mirror of
https://gitlab.gnome.org/GNOME/libsecret.git
synced 2024-12-22 04:38:55 +00:00
Complete Service instantiation, add mock service, and initial test
* Not yet tested, but compiles
This commit is contained in:
Stef Walter
Stef Walter
parent
8a6ba782db
commit
41e7cb387e
3
.gitignore
vendored
3
.gitignore
vendored
@ -2,6 +2,7 @@
|
||||
*.o
|
||||
*.la
|
||||
*.lo
|
||||
*.pyc
|
||||
.deps
|
||||
.cproject
|
||||
.libs
|
||||
@ -27,3 +28,5 @@ stamp*
|
||||
.cproject
|
||||
|
||||
/po/POTFILES
|
||||
|
||||
/library/tests/test-session
|
||||
|
||||
@ -165,10 +165,11 @@ on_service_ensure_session (GObject *source, GAsyncResult *result, gpointer user_
|
||||
GSimpleAsyncResult *res = G_SIMPLE_ASYNC_RESULT (user_data);
|
||||
GSecretItem *self = GSECRET_ITEM (g_async_result_get_source_object (user_data));
|
||||
GError *error = NULL;
|
||||
GCancellable *cancellable = NULL;
|
||||
const gchar *session_path;
|
||||
|
||||
session_path = gsecret_service_ensure_session_finish (self->pv->service,
|
||||
result, &error);
|
||||
session_path = _gsecret_service_ensure_session_finish (self->pv->service,
|
||||
result, &cancellable, &error);
|
||||
if (error != NULL) {
|
||||
g_simple_async_result_take_error (res, error);
|
||||
g_simple_async_result_complete (res);
|
||||
@ -177,11 +178,11 @@ on_service_ensure_session (GObject *source, GAsyncResult *result, gpointer user_
|
||||
g_assert (session_path != NULL && session_path[0] != '\0');
|
||||
g_dbus_proxy_call (G_DBUS_PROXY (self), "GetSecret",
|
||||
g_variant_new ("o", session_path),
|
||||
G_DBUS_CALL_FLAGS_NONE, -1,
|
||||
_gsecret_async_result_get_cancellable (res),
|
||||
G_DBUS_CALL_FLAGS_NONE, -1, cancellable,
|
||||
on_item_get_secret_ready, g_object_ref (res));
|
||||
}
|
||||
|
||||
g_clear_object (&cancellable);
|
||||
g_object_unref (res);
|
||||
}
|
||||
|
||||
|
||||
@ -20,22 +20,29 @@
|
||||
|
||||
G_BEGIN_DECLS
|
||||
|
||||
#define GSECRET_SERVICE_PATH "/org/freedesktop/secrets"
|
||||
|
||||
#define GSECRET_SERVICE_BUS_NAME "org.freedesktop.Secret.Service"
|
||||
|
||||
#define GSECRET_SERVICE_INTERFACE "org.freedesktop.Secret.Service"
|
||||
|
||||
#define GSECRET_COLLECTION_INTERFACE "org.freedesktop.Secret.Collection"
|
||||
|
||||
gchar * _gsecret_util_parent_path (const gchar *path);
|
||||
|
||||
GSecretService * _gsecret_service_bare_instance (GDBusConnection *connection,
|
||||
const gchar *bus_name);
|
||||
|
||||
GVariant * _gsecret_service_encode_secret (GSecretService *self,
|
||||
GSecretValue *value);
|
||||
|
||||
GSecretValue * _gsecret_service_decode_secret (GSecretService *service,
|
||||
GVariant *encoded);
|
||||
|
||||
GCancellable * _gsecret_async_result_get_cancellable (GSimpleAsyncResult *result);
|
||||
|
||||
void _gsecret_async_result_set_cancellable (GSimpleAsyncResult *result,
|
||||
GCancellable *cancellable);
|
||||
|
||||
gboolean _gsecret_async_result_propagate_cancelled (GSimpleAsyncResult *result);
|
||||
const gchar * _gsecret_service_ensure_session_finish (GSecretService *self,
|
||||
GAsyncResult *result,
|
||||
GCancellable **cancellable,
|
||||
GError **error);
|
||||
|
||||
G_END_DECLS
|
||||
|
||||
|
||||
@ -49,6 +49,11 @@ struct _GSecretServicePrivate {
|
||||
gpointer session;
|
||||
};
|
||||
|
||||
G_LOCK_DEFINE (service_instance);
|
||||
static gpointer service_instance = NULL;
|
||||
|
||||
G_DEFINE_TYPE (GSecretService, gsecret_service, G_TYPE_DBUS_PROXY);
|
||||
|
||||
static void
|
||||
gsecret_session_free (gpointer data)
|
||||
{
|
||||
@ -67,6 +72,90 @@ gsecret_session_free (gpointer data)
|
||||
g_free (session);
|
||||
}
|
||||
|
||||
static void
|
||||
gsecret_service_init (GSecretService *self)
|
||||
{
|
||||
self->pv = G_TYPE_INSTANCE_GET_PRIVATE (self, GSECRET_TYPE_SERVICE,
|
||||
GSecretServicePrivate);
|
||||
}
|
||||
|
||||
static void
|
||||
gsecret_service_finalize (GObject *obj)
|
||||
{
|
||||
GSecretService *self = GSECRET_SERVICE (obj);
|
||||
|
||||
gsecret_session_free (self->pv->session);
|
||||
|
||||
G_OBJECT_CLASS (gsecret_service_parent_class)->finalize (obj);
|
||||
}
|
||||
|
||||
static void
|
||||
gsecret_service_class_init (GSecretServiceClass *klass)
|
||||
{
|
||||
GObjectClass *object_class = G_OBJECT_CLASS (klass);
|
||||
|
||||
object_class->finalize = gsecret_service_finalize;
|
||||
|
||||
g_type_class_add_private (klass, sizeof (GSecretServicePrivate));
|
||||
}
|
||||
|
||||
static void
|
||||
on_service_instance_gone (gpointer user_data,
|
||||
GObject *where_the_object_was)
|
||||
{
|
||||
G_LOCK (service_instance);
|
||||
|
||||
g_assert (service_instance == where_the_object_was);
|
||||
service_instance = NULL;
|
||||
|
||||
G_UNLOCK (service_instance);
|
||||
}
|
||||
|
||||
GSecretService *
|
||||
_gsecret_service_bare_instance (GDBusConnection *connection,
|
||||
const gchar *bus_name)
|
||||
{
|
||||
GSecretService *service;
|
||||
GError *error = NULL;
|
||||
|
||||
g_return_val_if_fail (G_IS_DBUS_CONNECTION (connection), NULL);
|
||||
|
||||
/* Alternate bus name is only used for testing */
|
||||
if (bus_name == NULL)
|
||||
bus_name = GSECRET_SERVICE_BUS_NAME;
|
||||
|
||||
service = g_initable_new (GSECRET_TYPE_SERVICE, NULL, &error,
|
||||
"g-flags", G_DBUS_PROXY_FLAGS_DO_NOT_LOAD_PROPERTIES,
|
||||
"g-interface-info", NULL, /* TODO: */
|
||||
"g-name", bus_name,
|
||||
"g-connection", connection,
|
||||
"g-object-path", GSECRET_SERVICE_PATH,
|
||||
"g-interface-name", GSECRET_SERVICE_INTERFACE,
|
||||
NULL);
|
||||
|
||||
if (error != NULL) {
|
||||
g_warning ("couldn't create GSecretService object: %s", error->message);
|
||||
g_clear_error (&error);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
g_assert (GSECRET_IS_SERVICE (service));
|
||||
|
||||
G_LOCK (service_instance);
|
||||
|
||||
if (service_instance == NULL) {
|
||||
service_instance = service;
|
||||
g_object_weak_ref (G_OBJECT (service), on_service_instance_gone, NULL);
|
||||
} else {
|
||||
g_object_unref (service);
|
||||
service = g_object_ref (service_instance);
|
||||
}
|
||||
|
||||
G_UNLOCK (service_instance);
|
||||
|
||||
return service;
|
||||
}
|
||||
|
||||
#ifdef WITH_GCRYPT
|
||||
|
||||
static GVariant *
|
||||
@ -330,14 +419,17 @@ gsecret_service_ensure_session (GSecretService *self,
|
||||
}
|
||||
|
||||
const gchar *
|
||||
gsecret_service_ensure_session_finish (GSecretService *self,
|
||||
_gsecret_service_ensure_session_finish (GSecretService *self,
|
||||
GAsyncResult *result,
|
||||
GCancellable **cancellable,
|
||||
GError **error)
|
||||
{
|
||||
GSecretSession *session;
|
||||
OpenSessionClosure *closure;
|
||||
|
||||
g_return_val_if_fail (GSECRET_IS_SERVICE (self), NULL);
|
||||
g_return_val_if_fail (error == NULL || *error == NULL, NULL);
|
||||
g_return_val_if_fail (cancellable == NULL || *cancellable == NULL, NULL);
|
||||
|
||||
g_return_val_if_fail (g_simple_async_result_is_valid (result, G_OBJECT (self),
|
||||
gsecret_service_ensure_session), NULL);
|
||||
@ -345,12 +437,25 @@ gsecret_service_ensure_session_finish (GSecretService *self,
|
||||
if (g_simple_async_result_propagate_error (G_SIMPLE_ASYNC_RESULT (result), error))
|
||||
return NULL;
|
||||
|
||||
if (cancellable) {
|
||||
closure = g_simple_async_result_get_op_res_gpointer (G_SIMPLE_ASYNC_RESULT (result));
|
||||
*cancellable = closure->cancellable ? g_object_ref (closure->cancellable) : NULL;
|
||||
}
|
||||
|
||||
/* The session we have should never change once created */
|
||||
session = g_atomic_pointer_get (&self->pv->session);
|
||||
g_assert (session != NULL);
|
||||
return session->path;
|
||||
}
|
||||
|
||||
const gchar *
|
||||
gsecret_service_ensure_session_finish (GSecretService *self,
|
||||
GAsyncResult *result,
|
||||
GError **error)
|
||||
{
|
||||
return _gsecret_service_ensure_session_finish (self, result, NULL, error);
|
||||
}
|
||||
|
||||
const gchar *
|
||||
gsecret_service_ensure_session_sync (GSecretService *self,
|
||||
GCancellable *cancellable,
|
||||
|
||||
@ -13,9 +13,24 @@
|
||||
#include "config.h"
|
||||
|
||||
#include "gsecret-private.h"
|
||||
#include "gsecret-types.h"
|
||||
|
||||
#include <string.h>
|
||||
|
||||
GQuark
|
||||
gsecret_error_get_quark (void)
|
||||
{
|
||||
static volatile gsize initialized = 0;
|
||||
static GQuark quark = 0;
|
||||
|
||||
if (g_once_init_enter (&initialized)) {
|
||||
quark = g_quark_from_static_string ("gsecret-error");
|
||||
g_once_init_leave (&initialized, 1);
|
||||
}
|
||||
|
||||
return quark;
|
||||
}
|
||||
|
||||
gchar *
|
||||
_gsecret_util_parent_path (const gchar *path)
|
||||
{
|
||||
|
||||
@ -2,6 +2,7 @@ include $(top_srcdir)/Makefile.decl
|
||||
|
||||
INCLUDES = \
|
||||
-I$(top_srcdir)/library \
|
||||
-DSRCDIR="\"@abs_srcdir@\"" \
|
||||
$(NULL)
|
||||
|
||||
LDADD = \
|
||||
@ -10,7 +11,7 @@ LDADD = \
|
||||
$(NULL)
|
||||
|
||||
TEST_PROGS = \
|
||||
test-initial \
|
||||
test-session \
|
||||
$(NULL)
|
||||
|
||||
check_PROGRAMS = \
|
||||
|
||||
6
library/tests/mock-service-normal.py
Normal file
6
library/tests/mock-service-normal.py
Normal file
@ -0,0 +1,6 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
import mock
|
||||
|
||||
service = mock.SecretService()
|
||||
service.listen()
|
||||
@ -1,71 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
import os
|
||||
import sys
|
||||
import getopt
|
||||
import unittest
|
||||
|
||||
import dbus
|
||||
import dbus.service
|
||||
import dbus.glib
|
||||
import gobject
|
||||
|
||||
class NotSupported(dbus.exceptions.DBusException):
|
||||
def __init__(self, msg):
|
||||
dbus.exceptions.DBusException.__init__(self, msg, name="org.freedesktop.DBus.Error.NotSupported")
|
||||
|
||||
unique_identifier = 0
|
||||
def next_identifier():
|
||||
global unique_identifier
|
||||
unique_identifier += 1
|
||||
return unique_identifier
|
||||
|
||||
class SecretSession(dbus.service.Object):
|
||||
def __init__(self, service, key):
|
||||
global unique_identifier
|
||||
self.key = key
|
||||
self.path = "/org/freedesktop/secrets/sessions/%d" % next_identifier()
|
||||
dbus.service.Object.__init__(self, service.bus_name, self.path)
|
||||
|
||||
class SecretService(dbus.service.Object):
|
||||
|
||||
def __init__(self, name):
|
||||
self.bus_name = dbus.service.BusName(name, bus = dbus.SessionBus())
|
||||
dbus.service.Object.__init__(self, self.bus_name, '/org/freedesktop/secrets')
|
||||
self.sessions = { }
|
||||
|
||||
def listen(self):
|
||||
loop = gobject.MainLoop()
|
||||
loop.run()
|
||||
|
||||
@dbus.service.method('org.freedesktop.Secret.Service')
|
||||
def OpenSession(self, algorithm, input):
|
||||
assert type(algorithm) == dbus.String
|
||||
# assert type(input) == dbus.ByteArray
|
||||
|
||||
if algorithm == "plain":
|
||||
session = SecretSession(self, None)
|
||||
self.sessions[session.path] = session
|
||||
return (dbus.String("", variant_level=1), session)
|
||||
|
||||
elif algorithm == "dh-ietf1024-sha256-aes128-cbc-pkcs7":
|
||||
raise NotSupported("algorithm %s is not supported" % algorithm)
|
||||
|
||||
else:
|
||||
assert False, "algorithm %s not recognized" % algorithm
|
||||
|
||||
if __name__ == '__main__':
|
||||
try:
|
||||
opts, args = getopt.getopt(sys.argv[1:], "name", ["name="])
|
||||
except getopt.GetoptError, err:
|
||||
print str(err)
|
||||
sys.exit(2)
|
||||
name = 'org.freedesktop.Secret.MockService'
|
||||
for o, a in opts:
|
||||
if o in ("--name"):
|
||||
name = a
|
||||
else:
|
||||
assert False, "unhandled option"
|
||||
|
||||
myservice = SecretService(name)
|
||||
myservice.listen()
|
||||
1
library/tests/mock/__init__.py
Normal file
1
library/tests/mock/__init__.py
Normal file
@ -0,0 +1 @@
|
||||
from service import *
|
||||
656
library/tests/mock/aes.py
Normal file
656
library/tests/mock/aes.py
Normal file
@ -0,0 +1,656 @@
|
||||
#!/usr/bin/python
|
||||
#
|
||||
# aes.py: implements AES - Advanced Encryption Standard
|
||||
# from the SlowAES project, http://code.google.com/p/slowaes/
|
||||
#
|
||||
# Copyright (c) 2008 Josh Davis ( http://www.josh-davis.org ),
|
||||
# Alex Martelli ( http://www.aleax.it )
|
||||
#
|
||||
# Ported from C code written by Laurent Haan ( http://www.progressive-coding.com )
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0
|
||||
# http://www.apache.org/licenses/
|
||||
#
|
||||
import os
|
||||
import sys
|
||||
import math
|
||||
|
||||
def append_PKCS7_padding(s):
|
||||
"""return s padded to a multiple of 16-bytes by PKCS7 padding"""
|
||||
numpads = 16 - (len(s)%16)
|
||||
return s + numpads*chr(numpads)
|
||||
|
||||
def strip_PKCS7_padding(s):
|
||||
"""return s stripped of PKCS7 padding"""
|
||||
if len(s)%16 or not s:
|
||||
raise ValueError("String of len %d can't be PCKS7-padded" % len(s))
|
||||
numpads = ord(s[-1])
|
||||
if numpads > 16:
|
||||
raise ValueError("String ending with %r can't be PCKS7-padded" % s[-1])
|
||||
return s[:-numpads]
|
||||
|
||||
class AES(object):
|
||||
# valid key sizes
|
||||
keySize = dict(SIZE_128=16, SIZE_192=24, SIZE_256=32)
|
||||
|
||||
# Rijndael S-box
|
||||
sbox = [0x63, 0x7c, 0x77, 0x7b, 0xf2, 0x6b, 0x6f, 0xc5, 0x30, 0x01, 0x67,
|
||||
0x2b, 0xfe, 0xd7, 0xab, 0x76, 0xca, 0x82, 0xc9, 0x7d, 0xfa, 0x59,
|
||||
0x47, 0xf0, 0xad, 0xd4, 0xa2, 0xaf, 0x9c, 0xa4, 0x72, 0xc0, 0xb7,
|
||||
0xfd, 0x93, 0x26, 0x36, 0x3f, 0xf7, 0xcc, 0x34, 0xa5, 0xe5, 0xf1,
|
||||
0x71, 0xd8, 0x31, 0x15, 0x04, 0xc7, 0x23, 0xc3, 0x18, 0x96, 0x05,
|
||||
0x9a, 0x07, 0x12, 0x80, 0xe2, 0xeb, 0x27, 0xb2, 0x75, 0x09, 0x83,
|
||||
0x2c, 0x1a, 0x1b, 0x6e, 0x5a, 0xa0, 0x52, 0x3b, 0xd6, 0xb3, 0x29,
|
||||
0xe3, 0x2f, 0x84, 0x53, 0xd1, 0x00, 0xed, 0x20, 0xfc, 0xb1, 0x5b,
|
||||
0x6a, 0xcb, 0xbe, 0x39, 0x4a, 0x4c, 0x58, 0xcf, 0xd0, 0xef, 0xaa,
|
||||
0xfb, 0x43, 0x4d, 0x33, 0x85, 0x45, 0xf9, 0x02, 0x7f, 0x50, 0x3c,
|
||||
0x9f, 0xa8, 0x51, 0xa3, 0x40, 0x8f, 0x92, 0x9d, 0x38, 0xf5, 0xbc,
|
||||
0xb6, 0xda, 0x21, 0x10, 0xff, 0xf3, 0xd2, 0xcd, 0x0c, 0x13, 0xec,
|
||||
0x5f, 0x97, 0x44, 0x17, 0xc4, 0xa7, 0x7e, 0x3d, 0x64, 0x5d, 0x19,
|
||||
0x73, 0x60, 0x81, 0x4f, 0xdc, 0x22, 0x2a, 0x90, 0x88, 0x46, 0xee,
|
||||
0xb8, 0x14, 0xde, 0x5e, 0x0b, 0xdb, 0xe0, 0x32, 0x3a, 0x0a, 0x49,
|
||||
0x06, 0x24, 0x5c, 0xc2, 0xd3, 0xac, 0x62, 0x91, 0x95, 0xe4, 0x79,
|
||||
0xe7, 0xc8, 0x37, 0x6d, 0x8d, 0xd5, 0x4e, 0xa9, 0x6c, 0x56, 0xf4,
|
||||
0xea, 0x65, 0x7a, 0xae, 0x08, 0xba, 0x78, 0x25, 0x2e, 0x1c, 0xa6,
|
||||
0xb4, 0xc6, 0xe8, 0xdd, 0x74, 0x1f, 0x4b, 0xbd, 0x8b, 0x8a, 0x70,
|
||||
0x3e, 0xb5, 0x66, 0x48, 0x03, 0xf6, 0x0e, 0x61, 0x35, 0x57, 0xb9,
|
||||
0x86, 0xc1, 0x1d, 0x9e, 0xe1, 0xf8, 0x98, 0x11, 0x69, 0xd9, 0x8e,
|
||||
0x94, 0x9b, 0x1e, 0x87, 0xe9, 0xce, 0x55, 0x28, 0xdf, 0x8c, 0xa1,
|
||||
0x89, 0x0d, 0xbf, 0xe6, 0x42, 0x68, 0x41, 0x99, 0x2d, 0x0f, 0xb0,
|
||||
0x54, 0xbb, 0x16]
|
||||
|
||||
# Rijndael Inverted S-box
|
||||
rsbox = [0x52, 0x09, 0x6a, 0xd5, 0x30, 0x36, 0xa5, 0x38, 0xbf, 0x40, 0xa3,
|
||||
0x9e, 0x81, 0xf3, 0xd7, 0xfb , 0x7c, 0xe3, 0x39, 0x82, 0x9b, 0x2f,
|
||||
0xff, 0x87, 0x34, 0x8e, 0x43, 0x44, 0xc4, 0xde, 0xe9, 0xcb , 0x54,
|
||||
0x7b, 0x94, 0x32, 0xa6, 0xc2, 0x23, 0x3d, 0xee, 0x4c, 0x95, 0x0b,
|
||||
0x42, 0xfa, 0xc3, 0x4e , 0x08, 0x2e, 0xa1, 0x66, 0x28, 0xd9, 0x24,
|
||||
0xb2, 0x76, 0x5b, 0xa2, 0x49, 0x6d, 0x8b, 0xd1, 0x25 , 0x72, 0xf8,
|
||||
0xf6, 0x64, 0x86, 0x68, 0x98, 0x16, 0xd4, 0xa4, 0x5c, 0xcc, 0x5d,
|
||||
0x65, 0xb6, 0x92 , 0x6c, 0x70, 0x48, 0x50, 0xfd, 0xed, 0xb9, 0xda,
|
||||
0x5e, 0x15, 0x46, 0x57, 0xa7, 0x8d, 0x9d, 0x84 , 0x90, 0xd8, 0xab,
|
||||
0x00, 0x8c, 0xbc, 0xd3, 0x0a, 0xf7, 0xe4, 0x58, 0x05, 0xb8, 0xb3,
|
||||
0x45, 0x06 , 0xd0, 0x2c, 0x1e, 0x8f, 0xca, 0x3f, 0x0f, 0x02, 0xc1,
|
||||
0xaf, 0xbd, 0x03, 0x01, 0x13, 0x8a, 0x6b , 0x3a, 0x91, 0x11, 0x41,
|
||||
0x4f, 0x67, 0xdc, 0xea, 0x97, 0xf2, 0xcf, 0xce, 0xf0, 0xb4, 0xe6,
|
||||
0x73 , 0x96, 0xac, 0x74, 0x22, 0xe7, 0xad, 0x35, 0x85, 0xe2, 0xf9,
|
||||
0x37, 0xe8, 0x1c, 0x75, 0xdf, 0x6e , 0x47, 0xf1, 0x1a, 0x71, 0x1d,
|
||||
0x29, 0xc5, 0x89, 0x6f, 0xb7, 0x62, 0x0e, 0xaa, 0x18, 0xbe, 0x1b ,
|
||||
0xfc, 0x56, 0x3e, 0x4b, 0xc6, 0xd2, 0x79, 0x20, 0x9a, 0xdb, 0xc0,
|
||||
0xfe, 0x78, 0xcd, 0x5a, 0xf4 , 0x1f, 0xdd, 0xa8, 0x33, 0x88, 0x07,
|
||||
0xc7, 0x31, 0xb1, 0x12, 0x10, 0x59, 0x27, 0x80, 0xec, 0x5f , 0x60,
|
||||
0x51, 0x7f, 0xa9, 0x19, 0xb5, 0x4a, 0x0d, 0x2d, 0xe5, 0x7a, 0x9f,
|
||||
0x93, 0xc9, 0x9c, 0xef , 0xa0, 0xe0, 0x3b, 0x4d, 0xae, 0x2a, 0xf5,
|
||||
0xb0, 0xc8, 0xeb, 0xbb, 0x3c, 0x83, 0x53, 0x99, 0x61 , 0x17, 0x2b,
|
||||
0x04, 0x7e, 0xba, 0x77, 0xd6, 0x26, 0xe1, 0x69, 0x14, 0x63, 0x55,
|
||||
0x21, 0x0c, 0x7d]
|
||||
|
||||
def getSBoxValue(self,num):
|
||||
"""Retrieves a given S-Box Value"""
|
||||
return self.sbox[num]
|
||||
|
||||
def getSBoxInvert(self,num):
|
||||
"""Retrieves a given Inverted S-Box Value"""
|
||||
return self.rsbox[num]
|
||||
|
||||
def rotate(self, word):
|
||||
""" Rijndael's key schedule rotate operation.
|
||||
|
||||
Rotate a word eight bits to the left: eg, rotate(1d2c3a4f) == 2c3a4f1d
|
||||
Word is an char list of size 4 (32 bits overall).
|
||||
"""
|
||||
return word[1:] + word[:1]
|
||||
|
||||
# Rijndael Rcon
|
||||
Rcon = [0x8d, 0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1b, 0x36,
|
||||
0x6c, 0xd8, 0xab, 0x4d, 0x9a, 0x2f, 0x5e, 0xbc, 0x63, 0xc6, 0x97,
|
||||
0x35, 0x6a, 0xd4, 0xb3, 0x7d, 0xfa, 0xef, 0xc5, 0x91, 0x39, 0x72,
|
||||
0xe4, 0xd3, 0xbd, 0x61, 0xc2, 0x9f, 0x25, 0x4a, 0x94, 0x33, 0x66,
|
||||
0xcc, 0x83, 0x1d, 0x3a, 0x74, 0xe8, 0xcb, 0x8d, 0x01, 0x02, 0x04,
|
||||
0x08, 0x10, 0x20, 0x40, 0x80, 0x1b, 0x36, 0x6c, 0xd8, 0xab, 0x4d,
|
||||
0x9a, 0x2f, 0x5e, 0xbc, 0x63, 0xc6, 0x97, 0x35, 0x6a, 0xd4, 0xb3,
|
||||
0x7d, 0xfa, 0xef, 0xc5, 0x91, 0x39, 0x72, 0xe4, 0xd3, 0xbd, 0x61,
|
||||
0xc2, 0x9f, 0x25, 0x4a, 0x94, 0x33, 0x66, 0xcc, 0x83, 0x1d, 0x3a,
|
||||
0x74, 0xe8, 0xcb, 0x8d, 0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40,
|
||||
0x80, 0x1b, 0x36, 0x6c, 0xd8, 0xab, 0x4d, 0x9a, 0x2f, 0x5e, 0xbc,
|
||||
0x63, 0xc6, 0x97, 0x35, 0x6a, 0xd4, 0xb3, 0x7d, 0xfa, 0xef, 0xc5,
|
||||
0x91, 0x39, 0x72, 0xe4, 0xd3, 0xbd, 0x61, 0xc2, 0x9f, 0x25, 0x4a,
|
||||
0x94, 0x33, 0x66, 0xcc, 0x83, 0x1d, 0x3a, 0x74, 0xe8, 0xcb, 0x8d,
|
||||
0x01, 0x02, 0x04, 0x08, 0x10, 0x20, 0x40, 0x80, 0x1b, 0x36, 0x6c,
|
||||
0xd8, 0xab, 0x4d, 0x9a, 0x2f, 0x5e, 0xbc, 0x63, 0xc6, 0x97, 0x35,
|
||||
0x6a, 0xd4, 0xb3, 0x7d, 0xfa, 0xef, 0xc5, 0x91, 0x39, 0x72, 0xe4,
|
||||
0xd3, 0xbd, 0x61, 0xc2, 0x9f, 0x25, 0x4a, 0x94, 0x33, 0x66, 0xcc,
|
||||
0x83, 0x1d, 0x3a, 0x74, 0xe8, 0xcb, 0x8d, 0x01, 0x02, 0x04, 0x08,
|
||||
0x10, 0x20, 0x40, 0x80, 0x1b, 0x36, 0x6c, 0xd8, 0xab, 0x4d, 0x9a,
|
||||
0x2f, 0x5e, 0xbc, 0x63, 0xc6, 0x97, 0x35, 0x6a, 0xd4, 0xb3, 0x7d,
|
||||
0xfa, 0xef, 0xc5, 0x91, 0x39, 0x72, 0xe4, 0xd3, 0xbd, 0x61, 0xc2,
|
||||
0x9f, 0x25, 0x4a, 0x94, 0x33, 0x66, 0xcc, 0x83, 0x1d, 0x3a, 0x74,
|
||||
0xe8, 0xcb ]
|
||||
|
||||
def getRconValue(self, num):
|
||||
"""Retrieves a given Rcon Value"""
|
||||
return self.Rcon[num]
|
||||
|
||||
def core(self, word, iteration):
|
||||
"""Key schedule core."""
|
||||
# rotate the 32-bit word 8 bits to the left
|
||||
word = self.rotate(word)
|
||||
# apply S-Box substitution on all 4 parts of the 32-bit word
|
||||
for i in range(4):
|
||||
word[i] = self.getSBoxValue(word[i])
|
||||
# XOR the output of the rcon operation with i to the first part
|
||||
# (leftmost) only
|
||||
word[0] = word[0] ^ self.getRconValue(iteration)
|
||||
return word
|
||||
|
||||
def expandKey(self, key, size, expandedKeySize):
|
||||
"""Rijndael's key expansion.
|
||||
|
||||
Expands an 128,192,256 key into an 176,208,240 bytes key
|
||||
|
||||
expandedKey is a char list of large enough size,
|
||||
key is the non-expanded key.
|
||||
"""
|
||||
# current expanded keySize, in bytes
|
||||
currentSize = 0
|
||||
rconIteration = 1
|
||||
expandedKey = [0] * expandedKeySize
|
||||
|
||||
# set the 16, 24, 32 bytes of the expanded key to the input key
|
||||
for j in range(size):
|
||||
expandedKey[j] = key[j]
|
||||
currentSize += size
|
||||
|
||||
while currentSize < expandedKeySize:
|
||||
# assign the previous 4 bytes to the temporary value t
|
||||
t = expandedKey[currentSize-4:currentSize]
|
||||
|
||||
# every 16,24,32 bytes we apply the core schedule to t
|
||||
# and increment rconIteration afterwards
|
||||
if currentSize % size == 0:
|
||||
t = self.core(t, rconIteration)
|
||||
rconIteration += 1
|
||||
# For 256-bit keys, we add an extra sbox to the calculation
|
||||
if size == self.keySize["SIZE_256"] and ((currentSize % size) == 16):
|
||||
for l in range(4): t[l] = self.getSBoxValue(t[l])
|
||||
|
||||
# We XOR t with the four-byte block 16,24,32 bytes before the new
|
||||
# expanded key. This becomes the next four bytes in the expanded
|
||||
# key.
|
||||
for m in range(4):
|
||||
expandedKey[currentSize] = expandedKey[currentSize - size] ^ \
|
||||
t[m]
|
||||
currentSize += 1
|
||||
|
||||
return expandedKey
|
||||
|
||||
def addRoundKey(self, state, roundKey):
|
||||
"""Adds (XORs) the round key to the state."""
|
||||
for i in range(16):
|
||||
state[i] ^= roundKey[i]
|
||||
return state
|
||||
|
||||
def createRoundKey(self, expandedKey, roundKeyPointer):
|
||||
"""Create a round key.
|
||||
Creates a round key from the given expanded key and the
|
||||
position within the expanded key.
|
||||
"""
|
||||
roundKey = [0] * 16
|
||||
for i in range(4):
|
||||
for j in range(4):
|
||||
roundKey[j*4+i] = expandedKey[roundKeyPointer + i*4 + j]
|
||||
return roundKey
|
||||
|
||||
def galois_multiplication(self, a, b):
|
||||
"""Galois multiplication of 8 bit characters a and b."""
|
||||
p = 0
|
||||
for counter in range(8):
|
||||
if b & 1: p ^= a
|
||||
hi_bit_set = a & 0x80
|
||||
a <<= 1
|
||||
# keep a 8 bit
|
||||
a &= 0xFF
|
||||
if hi_bit_set:
|
||||
a ^= 0x1b
|
||||
b >>= 1
|
||||
return p
|
||||
|
||||
#
|
||||
# substitute all the values from the state with the value in the SBox
|
||||
# using the state value as index for the SBox
|
||||
#
|
||||
def subBytes(self, state, isInv):
|
||||
if isInv: getter = self.getSBoxInvert
|
||||
else: getter = self.getSBoxValue
|
||||
for i in range(16): state[i] = getter(state[i])
|
||||
return state
|
||||
|
||||
# iterate over the 4 rows and call shiftRow() with that row
|
||||
def shiftRows(self, state, isInv):
|
||||
for i in range(4):
|
||||
state = self.shiftRow(state, i*4, i, isInv)
|
||||
return state
|
||||
|
||||
# each iteration shifts the row to the left by 1
|
||||
def shiftRow(self, state, statePointer, nbr, isInv):
|
||||
for i in range(nbr):
|
||||
if isInv:
|
||||
state[statePointer:statePointer+4] = \
|
||||
state[statePointer+3:statePointer+4] + \
|
||||
state[statePointer:statePointer+3]
|
||||
else:
|
||||
state[statePointer:statePointer+4] = \
|
||||
state[statePointer+1:statePointer+4] + \
|
||||
state[statePointer:statePointer+1]
|
||||
return state
|
||||
|
||||
# galois multiplication of the 4x4 matrix
|
||||
def mixColumns(self, state, isInv):
|
||||
# iterate over the 4 columns
|
||||
for i in range(4):
|
||||
# construct one column by slicing over the 4 rows
|
||||
column = state[i:i+16:4]
|
||||
# apply the mixColumn on one column
|
||||
column = self.mixColumn(column, isInv)
|
||||
# put the values back into the state
|
||||
state[i:i+16:4] = column
|
||||
|
||||
return state
|
||||
|
||||
# galois multiplication of 1 column of the 4x4 matrix
|
||||
def mixColumn(self, column, isInv):
|
||||
if isInv: mult = [14, 9, 13, 11]
|
||||
else: mult = [2, 1, 1, 3]
|
||||
cpy = list(column)
|
||||
g = self.galois_multiplication
|
||||
|
||||
column[0] = g(cpy[0], mult[0]) ^ g(cpy[3], mult[1]) ^ \
|
||||
g(cpy[2], mult[2]) ^ g(cpy[1], mult[3])
|
||||
column[1] = g(cpy[1], mult[0]) ^ g(cpy[0], mult[1]) ^ \
|
||||
g(cpy[3], mult[2]) ^ g(cpy[2], mult[3])
|
||||
column[2] = g(cpy[2], mult[0]) ^ g(cpy[1], mult[1]) ^ \
|
||||
g(cpy[0], mult[2]) ^ g(cpy[3], mult[3])
|
||||
column[3] = g(cpy[3], mult[0]) ^ g(cpy[2], mult[1]) ^ \
|
||||
g(cpy[1], mult[2]) ^ g(cpy[0], mult[3])
|
||||
return column
|
||||
|
||||
# applies the 4 operations of the forward round in sequence
|
||||
def aes_round(self, state, roundKey):
|
||||
state = self.subBytes(state, False)
|
||||
state = self.shiftRows(state, False)
|
||||
state = self.mixColumns(state, False)
|
||||
state = self.addRoundKey(state, roundKey)
|
||||
return state
|
||||
|
||||
# applies the 4 operations of the inverse round in sequence
|
||||
def aes_invRound(self, state, roundKey):
|
||||
state = self.shiftRows(state, True)
|
||||
state = self.subBytes(state, True)
|
||||
state = self.addRoundKey(state, roundKey)
|
||||
state = self.mixColumns(state, True)
|
||||
return state
|
||||
|
||||
# Perform the initial operations, the standard round, and the final
|
||||
# operations of the forward aes, creating a round key for each round
|
||||
def aes_main(self, state, expandedKey, nbrRounds):
|
||||
state = self.addRoundKey(state, self.createRoundKey(expandedKey, 0))
|
||||
i = 1
|
||||
while i < nbrRounds:
|
||||
state = self.aes_round(state,
|
||||
self.createRoundKey(expandedKey, 16*i))
|
||||
i += 1
|
||||
state = self.subBytes(state, False)
|
||||
state = self.shiftRows(state, False)
|
||||
state = self.addRoundKey(state,
|
||||
self.createRoundKey(expandedKey, 16*nbrRounds))
|
||||
return state
|
||||
|
||||
# Perform the initial operations, the standard round, and the final
|
||||
# operations of the inverse aes, creating a round key for each round
|
||||
def aes_invMain(self, state, expandedKey, nbrRounds):
|
||||
state = self.addRoundKey(state,
|
||||
self.createRoundKey(expandedKey, 16*nbrRounds))
|
||||
i = nbrRounds - 1
|
||||
while i > 0:
|
||||
state = self.aes_invRound(state,
|
||||
self.createRoundKey(expandedKey, 16*i))
|
||||
i -= 1
|
||||
state = self.shiftRows(state, True)
|
||||
state = self.subBytes(state, True)
|
||||
state = self.addRoundKey(state, self.createRoundKey(expandedKey, 0))
|
||||
return state
|
||||
|
||||
# encrypts a 128 bit input block against the given key of size specified
|
||||
def encrypt(self, iput, key, size):
|
||||
output = [0] * 16
|
||||
# the number of rounds
|
||||
nbrRounds = 0
|
||||
# the 128 bit block to encode
|
||||
block = [0] * 16
|
||||
# set the number of rounds
|
||||
if size == self.keySize["SIZE_128"]: nbrRounds = 10
|
||||
elif size == self.keySize["SIZE_192"]: nbrRounds = 12
|
||||
elif size == self.keySize["SIZE_256"]: nbrRounds = 14
|
||||
else: return None
|
||||
|
||||
# the expanded keySize
|
||||
expandedKeySize = 16*(nbrRounds+1)
|
||||
|
||||
# Set the block values, for the block:
|
||||
# a0,0 a0,1 a0,2 a0,3
|
||||
# a1,0 a1,1 a1,2 a1,3
|
||||
# a2,0 a2,1 a2,2 a2,3
|
||||
# a3,0 a3,1 a3,2 a3,3
|
||||
# the mapping order is a0,0 a1,0 a2,0 a3,0 a0,1 a1,1 ... a2,3 a3,3
|
||||
#
|
||||
# iterate over the columns
|
||||
for i in range(4):
|
||||
# iterate over the rows
|
||||
for j in range(4):
|
||||
block[(i+(j*4))] = iput[(i*4)+j]
|
||||
|
||||
# expand the key into an 176, 208, 240 bytes key
|
||||
# the expanded key
|
||||
expandedKey = self.expandKey(key, size, expandedKeySize)
|
||||
|
||||
# encrypt the block using the expandedKey
|
||||
block = self.aes_main(block, expandedKey, nbrRounds)
|
||||
|
||||
# unmap the block again into the output
|
||||
for k in range(4):
|
||||
# iterate over the rows
|
||||
for l in range(4):
|
||||
output[(k*4)+l] = block[(k+(l*4))]
|
||||
return output
|
||||
|
||||
# decrypts a 128 bit input block against the given key of size specified
|
||||
def decrypt(self, iput, key, size):
|
||||
output = [0] * 16
|
||||
# the number of rounds
|
||||
nbrRounds = 0
|
||||
# the 128 bit block to decode
|
||||
block = [0] * 16
|
||||
# set the number of rounds
|
||||
if size == self.keySize["SIZE_128"]: nbrRounds = 10
|
||||
elif size == self.keySize["SIZE_192"]: nbrRounds = 12
|
||||
elif size == self.keySize["SIZE_256"]: nbrRounds = 14
|
||||
else: return None
|
||||
|
||||
# the expanded keySize
|
||||
expandedKeySize = 16*(nbrRounds+1)
|
||||
|
||||
# Set the block values, for the block:
|
||||
# a0,0 a0,1 a0,2 a0,3
|
||||
# a1,0 a1,1 a1,2 a1,3
|
||||
# a2,0 a2,1 a2,2 a2,3
|
||||
# a3,0 a3,1 a3,2 a3,3
|
||||
# the mapping order is a0,0 a1,0 a2,0 a3,0 a0,1 a1,1 ... a2,3 a3,3
|
||||
|
||||
# iterate over the columns
|
||||
for i in range(4):
|
||||
# iterate over the rows
|
||||
for j in range(4):
|
||||
block[(i+(j*4))] = iput[(i*4)+j]
|
||||
# expand the key into an 176, 208, 240 bytes key
|
||||
expandedKey = self.expandKey(key, size, expandedKeySize)
|
||||
# decrypt the block using the expandedKey
|
||||
block = self.aes_invMain(block, expandedKey, nbrRounds)
|
||||
# unmap the block again into the output
|
||||
for k in range(4):
|
||||
# iterate over the rows
|
||||
for l in range(4):
|
||||
output[(k*4)+l] = block[(k+(l*4))]
|
||||
return output
|
||||
|
||||
|
||||
class AESModeOfOperation(object):
|
||||
|
||||
aes = AES()
|
||||
|
||||
# structure of supported modes of operation
|
||||
modeOfOperation = dict(OFB=0, CFB=1, CBC=2)
|
||||
|
||||
# converts a 16 character string into a number array
|
||||
def convertString(self, string, start, end, mode):
|
||||
if end - start > 16: end = start + 16
|
||||
if mode == self.modeOfOperation["CBC"]: ar = [0] * 16
|
||||
else: ar = []
|
||||
|
||||
i = start
|
||||
j = 0
|
||||
while len(ar) < end - start:
|
||||
ar.append(0)
|
||||
while i < end:
|
||||
ar[j] = ord(string[i])
|
||||
j += 1
|
||||
i += 1
|
||||
return ar
|
||||
|
||||
# Mode of Operation Encryption
|
||||
# stringIn - Input String
|
||||
# mode - mode of type modeOfOperation
|
||||
# hexKey - a hex key of the bit length size
|
||||
# size - the bit length of the key
|
||||
# hexIV - the 128 bit hex Initilization Vector
|
||||
def encrypt(self, stringIn, mode, key, size, IV):
|
||||
if len(key) % size:
|
||||
return None
|
||||
if len(IV) % 16:
|
||||
return None
|
||||
# the AES input/output
|
||||
plaintext = []
|
||||
iput = [0] * 16
|
||||
output = []
|
||||
ciphertext = [0] * 16
|
||||
# the output cipher string
|
||||
cipherOut = []
|
||||
# char firstRound
|
||||
firstRound = True
|
||||
if stringIn != None:
|
||||
for j in range(int(math.ceil(float(len(stringIn))/16))):
|
||||
start = j*16
|
||||
end = j*16+16
|
||||
if end > len(stringIn):
|
||||
end = len(stringIn)
|
||||
plaintext = self.convertString(stringIn, start, end, mode)
|
||||
# print 'PT@%s:%s' % (j, plaintext)
|
||||
if mode == self.modeOfOperation["CFB"]:
|
||||
if firstRound:
|
||||
output = self.aes.encrypt(IV, key, size)
|
||||
firstRound = False
|
||||
else:
|
||||
output = self.aes.encrypt(iput, key, size)
|
||||
for i in range(16):
|
||||
if len(plaintext)-1 < i:
|
||||
ciphertext[i] = 0 ^ output[i]
|
||||
elif len(output)-1 < i:
|
||||
ciphertext[i] = plaintext[i] ^ 0
|
||||
elif len(plaintext)-1 < i and len(output) < i:
|
||||
ciphertext[i] = 0 ^ 0
|
||||
else:
|
||||
ciphertext[i] = plaintext[i] ^ output[i]
|
||||
for k in range(end-start):
|
||||
cipherOut.append(ciphertext[k])
|
||||
iput = ciphertext
|
||||
elif mode == self.modeOfOperation["OFB"]:
|
||||
if firstRound:
|
||||
output = self.aes.encrypt(IV, key, size)
|
||||
firstRound = False
|
||||
else:
|
||||
output = self.aes.encrypt(iput, key, size)
|
||||
for i in range(16):
|
||||
if len(plaintext)-1 < i:
|
||||
ciphertext[i] = 0 ^ output[i]
|
||||
elif len(output)-1 < i:
|
||||
ciphertext[i] = plaintext[i] ^ 0
|
||||
elif len(plaintext)-1 < i and len(output) < i:
|
||||
ciphertext[i] = 0 ^ 0
|
||||
else:
|
||||
ciphertext[i] = plaintext[i] ^ output[i]
|
||||
for k in range(end-start):
|
||||
cipherOut.append(ciphertext[k])
|
||||
iput = output
|
||||
elif mode == self.modeOfOperation["CBC"]:
|
||||
for i in range(16):
|
||||
if firstRound:
|
||||
iput[i] = plaintext[i] ^ IV[i]
|
||||
else:
|
||||
iput[i] = plaintext[i] ^ ciphertext[i]
|
||||
# print 'IP@%s:%s' % (j, iput)
|
||||
firstRound = False
|
||||
ciphertext = self.aes.encrypt(iput, key, size)
|
||||
# always 16 bytes because of the padding for CBC
|
||||
for k in range(16):
|
||||
cipherOut.append(ciphertext[k])
|
||||
return mode, len(stringIn), cipherOut
|
||||
|
||||
# Mode of Operation Decryption
|
||||
# cipherIn - Encrypted String
|
||||
# originalsize - The unencrypted string length - required for CBC
|
||||
# mode - mode of type modeOfOperation
|
||||
# key - a number array of the bit length size
|
||||
# size - the bit length of the key
|
||||
# IV - the 128 bit number array Initilization Vector
|
||||
def decrypt(self, cipherIn, originalsize, mode, key, size, IV):
|
||||
# cipherIn = unescCtrlChars(cipherIn)
|
||||
if len(key) % size:
|
||||
return None
|
||||
if len(IV) % 16:
|
||||
return None
|
||||
# the AES input/output
|
||||
ciphertext = []
|
||||
iput = []
|
||||
output = []
|
||||
plaintext = [0] * 16
|
||||
# the output plain text string
|
||||
stringOut = ''
|
||||
# char firstRound
|
||||
firstRound = True
|
||||
if cipherIn != None:
|
||||
for j in range(int(math.ceil(float(len(cipherIn))/16))):
|
||||
start = j*16
|
||||
end = j*16+16
|
||||
if j*16+16 > len(cipherIn):
|
||||
end = len(cipherIn)
|
||||
ciphertext = cipherIn[start:end]
|
||||
if mode == self.modeOfOperation["CFB"]:
|
||||
if firstRound:
|
||||
output = self.aes.encrypt(IV, key, size)
|
||||
firstRound = False
|
||||
else:
|
||||
output = self.aes.encrypt(iput, key, size)
|
||||
for i in range(16):
|
||||
if len(output)-1 < i:
|
||||
plaintext[i] = 0 ^ ciphertext[i]
|
||||
elif len(ciphertext)-1 < i:
|
||||
plaintext[i] = output[i] ^ 0
|
||||
elif len(output)-1 < i and len(ciphertext) < i:
|
||||
plaintext[i] = 0 ^ 0
|
||||
else:
|
||||
plaintext[i] = output[i] ^ ciphertext[i]
|
||||
for k in range(end-start):
|
||||
stringOut += chr(plaintext[k])
|
||||
iput = ciphertext
|
||||
elif mode == self.modeOfOperation["OFB"]:
|
||||
if firstRound:
|
||||
output = self.aes.encrypt(IV, key, size)
|
||||
firstRound = False
|
||||
else:
|
||||
output = self.aes.encrypt(iput, key, size)
|
||||
for i in range(16):
|
||||
if len(output)-1 < i:
|
||||
plaintext[i] = 0 ^ ciphertext[i]
|
||||
elif len(ciphertext)-1 < i:
|
||||
plaintext[i] = output[i] ^ 0
|
||||
elif len(output)-1 < i and len(ciphertext) < i:
|
||||
plaintext[i] = 0 ^ 0
|
||||
else:
|
||||
plaintext[i] = output[i] ^ ciphertext[i]
|
||||
for k in range(end-start):
|
||||
stringOut += chr(plaintext[k])
|
||||
iput = output
|
||||
elif mode == self.modeOfOperation["CBC"]:
|
||||
output = self.aes.decrypt(ciphertext, key, size)
|
||||
for i in range(16):
|
||||
if firstRound:
|
||||
plaintext[i] = IV[i] ^ output[i]
|
||||
else:
|
||||
plaintext[i] = iput[i] ^ output[i]
|
||||
firstRound = False
|
||||
if originalsize is not None and originalsize < end:
|
||||
for k in range(originalsize-start):
|
||||
stringOut += chr(plaintext[k])
|
||||
else:
|
||||
for k in range(end-start):
|
||||
stringOut += chr(plaintext[k])
|
||||
iput = ciphertext
|
||||
return stringOut
|
||||
|
||||
|
||||
def encryptData(key, data, mode=AESModeOfOperation.modeOfOperation["CBC"]):
|
||||
"""encrypt `data` using `key`
|
||||
|
||||
`key` should be a string of bytes.
|
||||
|
||||
returned cipher is a string of bytes prepended with the initialization
|
||||
vector.
|
||||
|
||||
"""
|
||||
key = map(ord, key)
|
||||
if mode == AESModeOfOperation.modeOfOperation["CBC"]:
|
||||
data = append_PKCS7_padding(data)
|
||||
keysize = len(key)
|
||||
assert keysize in AES.keySize.values(), 'invalid key size: %s' % keysize
|
||||
# create a new iv using random data
|
||||
iv = [ord(i) for i in os.urandom(16)]
|
||||
moo = AESModeOfOperation()
|
||||
(mode, length, ciph) = moo.encrypt(data, mode, key, keysize, iv)
|
||||
# With padding, the original length does not need to be known. It's a bad
|
||||
# idea to store the original message length.
|
||||
# prepend the iv.
|
||||
return ''.join(map(chr, iv)) + ''.join(map(chr, ciph))
|
||||
|
||||
def decryptData(key, data, mode=AESModeOfOperation.modeOfOperation["CBC"]):
|
||||
"""decrypt `data` using `key`
|
||||
|
||||
`key` should be a string of bytes.
|
||||
|
||||
`data` should have the initialization vector prepended as a string of
|
||||
ordinal values.
|
||||
|
||||
"""
|
||||
|
||||
key = map(ord, key)
|
||||
keysize = len(key)
|
||||
assert keysize in AES.keySize.values(), 'invalid key size: %s' % keysize
|
||||
# iv is first 16 bytes
|
||||
iv = map(ord, data[:16])
|
||||
data = map(ord, data[16:])
|
||||
moo = AESModeOfOperation()
|
||||
decr = moo.decrypt(data, None, mode, key, keysize, iv)
|
||||
if mode == AESModeOfOperation.modeOfOperation["CBC"]:
|
||||
decr = strip_PKCS7_padding(decr)
|
||||
return decr
|
||||
|
||||
def generateRandomKey(keysize):
|
||||
"""Generates a key from random data of length `keysize`.
|
||||
|
||||
The returned key is a string of bytes.
|
||||
|
||||
"""
|
||||
if keysize not in (16, 24, 32):
|
||||
emsg = 'Invalid keysize, %s. Should be one of (16, 24, 32).'
|
||||
raise ValueError, emsg % keysize
|
||||
return os.urandom(keysize)
|
||||
|
||||
if __name__ == "__main__":
|
||||
moo = AESModeOfOperation()
|
||||
cleartext = "This is a test!"
|
||||
cypherkey = [143,194,34,208,145,203,230,143,177,246,97,206,145,92,255,84]
|
||||
iv = [103,35,148,239,76,213,47,118,255,222,123,176,106,134,98,92]
|
||||
mode, orig_len, ciph = moo.encrypt(cleartext, moo.modeOfOperation["CBC"],
|
||||
cypherkey, moo.aes.keySize["SIZE_128"], iv)
|
||||
print 'm=%s, ol=%s (%s), ciph=%s' % (mode, orig_len, len(cleartext), ciph)
|
||||
decr = moo.decrypt(ciph, orig_len, mode, cypherkey,
|
||||
moo.aes.keySize["SIZE_128"], iv)
|
||||
print decr
|
||||
80
library/tests/mock/dh.py
Normal file
80
library/tests/mock/dh.py
Normal file
@ -0,0 +1,80 @@
|
||||
|
||||
# WARNING: This is for use in mock objects during testing, and NOT
|
||||
# cryptographically secure or performant.
|
||||
|
||||
#
|
||||
# Copyright 2011 Stef Walter
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Lesser General Public License as published
|
||||
# by the Free Software Foundation; either version 2 of the licence or (at
|
||||
# your option) any later version.
|
||||
#
|
||||
# See the included COPYING file for more information.
|
||||
#
|
||||
|
||||
#
|
||||
# Some utility functions from tlslite which is public domain
|
||||
# Written by Trevor Perrin <trevp at trevp.net>
|
||||
# http://trevp.net/tlslite/
|
||||
#
|
||||
|
||||
import math
|
||||
import random
|
||||
|
||||
PRIME = '\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xC9\x0F\xDA\xA2\x21\x68\xC2\x34\xC4' \
|
||||
'\xC6\x62\x8B\x80\xDC\x1C\xD1\x29\x02\x4E\x08\x8A\x67\xCC\x74\x02\x0B\xBE' \
|
||||
'\xA6\x3B\x13\x9B\x22\x51\x4A\x08\x79\x8E\x34\x04\xDD\xEF\x95\x19\xB3\xCD' \
|
||||
'\x3A\x43\x1B\x30\x2B\x0A\x6D\xF2\x5F\x14\x37\x4F\xE1\x35\x6D\x6D\x51\xC2' \
|
||||
'\x45\xE4\x85\xB5\x76\x62\x5E\x7E\xC6\xF4\x4C\x42\xE9\xA6\x37\xED\x6B\x0B' \
|
||||
'\xFF\x5C\xB6\xF4\x06\xB7\xED\xEE\x38\x6B\xFB\x5A\x89\x9F\xA5\xAE\x9F\x24' \
|
||||
'\x11\x7C\x4B\x1F\xE6\x49\x28\x66\x51\xEC\xE6\x53\x81\xFF\xFF\xFF\xFF\xFF' \
|
||||
'\xFF\xFF\xFF'
|
||||
|
||||
def num_bits(number):
|
||||
if number == 0:
|
||||
return 0
|
||||
s = "%x" % number
|
||||
return ((len(s)-1)*4) + \
|
||||
{'0':0, '1':1, '2':2, '3':2,
|
||||
'4':3, '5':3, '6':3, '7':3,
|
||||
'8':4, '9':4, 'a':4, 'b':4,
|
||||
'c':4, 'd':4, 'e':4, 'f':4,
|
||||
}[s[0]]
|
||||
|
||||
def num_bytes(number):
|
||||
if number == 0:
|
||||
return 0
|
||||
bits = num_bits(number)
|
||||
return int(math.ceil(bits / 8.0))
|
||||
|
||||
def bytes_to_number(data):
|
||||
number = 0L
|
||||
multiplier = 1L
|
||||
for count in range(len(data) - 1, -1, -1):
|
||||
number += multiplier * ord(data[count])
|
||||
multiplier *= 256
|
||||
return number
|
||||
|
||||
def number_to_bytes(number):
|
||||
n_data = num_bytes(number)
|
||||
data = ['' for i in range(0, n_data)]
|
||||
for count in range(n_data - 1, -1, -1):
|
||||
data[count] = chr(number % 256)
|
||||
number >>= 8
|
||||
return "".join(data)
|
||||
|
||||
def generate_pair():
|
||||
prime = bytes_to_number (PRIME)
|
||||
base = 2
|
||||
bits = num_bits(prime)
|
||||
privat = 0
|
||||
while privat == 0:
|
||||
privat = random.getrandbits(bits - 1)
|
||||
publi = pow(base, privat, prime)
|
||||
return (privat, publi)
|
||||
|
||||
def derive_key(privat, peer):
|
||||
prime = bytes_to_number (PRIME)
|
||||
key = pow(peer, privat, prime)
|
||||
return number_to_bytes(key)
|
||||
86
library/tests/mock/hkdf.py
Normal file
86
library/tests/mock/hkdf.py
Normal file
@ -0,0 +1,86 @@
|
||||
|
||||
# Upstream Author: Zooko O'Whielacronx <zooko@zooko.com>
|
||||
#
|
||||
# Copyright:
|
||||
#
|
||||
# You may use this package under the GNU General Public License, version
|
||||
# 2 or, at your option, any later version. You may use this package
|
||||
# under the Transitive Grace Period Public Licence, version 1.0 or, at
|
||||
# your option, any later version. (You may choose to use this package
|
||||
# under the terms of either licence, at your option.) See the file
|
||||
# COPYING.GPL for the terms of the GNU General Public License, version 2.
|
||||
# See the file COPYING.TGPPL.html for the terms of the Transitive Grace
|
||||
# Period Public Licence, version 1.0.
|
||||
#
|
||||
# The following licensing text applies to a subset of the Crypto++ source code
|
||||
# which is included in the pycryptopp source tree under the "embeddedcryptopp"
|
||||
# subdirectory. That embedded subset of the Crypto++ source code is not used
|
||||
# when pycryptopp is built for Debian -- instead the --disable-embedded-cryptopp
|
||||
# option to "setup.py build" is used to for pycryptopp to build against the
|
||||
# system libcryptopp.
|
||||
|
||||
import hashlib, hmac
|
||||
import math
|
||||
from binascii import a2b_hex, b2a_hex
|
||||
|
||||
class HKDF(object):
|
||||
def __init__(self, ikm, L, salt=None, info="", digestmod = None):
|
||||
self.ikm = ikm
|
||||
self.keylen = L
|
||||
|
||||
if digestmod is None:
|
||||
digestmod = hashlib.sha256
|
||||
|
||||
if callable(digestmod):
|
||||
self.digest_cons = digestmod
|
||||
else:
|
||||
self.digest_cons = lambda d='':digestmod.new(d)
|
||||
self.hashlen = len(self.digest_cons().digest())
|
||||
|
||||
if salt is None:
|
||||
self.salt = chr(0)*(self.hashlen)
|
||||
else:
|
||||
self.salt = salt
|
||||
|
||||
self.info = info
|
||||
|
||||
#extract PRK
|
||||
def extract(self):
|
||||
h = hmac.new(self.salt, self.ikm, self.digest_cons)
|
||||
self.prk = h.digest()
|
||||
return self.prk
|
||||
|
||||
#expand PRK
|
||||
def expand(self):
|
||||
N = math.ceil(float(self.keylen)/self.hashlen)
|
||||
T = ""
|
||||
temp = ""
|
||||
i=0x01
|
||||
'''while len(T)<2*self.keylen :
|
||||
msg = temp
|
||||
msg += self.info
|
||||
msg += b2a_hex(chr(i))
|
||||
h = hmac.new(self.prk, a2b_hex(msg), self.digest_cons)
|
||||
temp = b2a_hex(h.digest())
|
||||
i += 1
|
||||
T += temp
|
||||
'''
|
||||
while len(T)<self.keylen :
|
||||
msg = temp
|
||||
msg += self.info
|
||||
msg += chr(i)
|
||||
h = hmac.new(self.prk, msg, self.digest_cons)
|
||||
temp = h.digest()
|
||||
i += 1
|
||||
T += temp
|
||||
|
||||
self.okm = T[0:self.keylen]
|
||||
return self.okm
|
||||
|
||||
def new(ikm, L, salt=None, info="", digestmod = None):
|
||||
return HKDF(ikm, L,salt,info,digestmod)
|
||||
|
||||
def hkdf(ikm, length, salt=None, info=""):
|
||||
hk = HKDF(ikm, length ,salt,info)
|
||||
computedprk = hk.extract()
|
||||
return hk.expand()
|
||||
126
library/tests/mock/service.py
Normal file
126
library/tests/mock/service.py
Normal file
@ -0,0 +1,126 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
#
|
||||
# Copyright 2011 Stef Walter
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Lesser General Public License as published
|
||||
# by the Free Software Foundation; either version 2 of the licence or (at
|
||||
# your option) any later version.
|
||||
#
|
||||
# See the included COPYING file for more information.
|
||||
#
|
||||
|
||||
import os
|
||||
import sys
|
||||
import getopt
|
||||
import unittest
|
||||
|
||||
import aes
|
||||
import dh
|
||||
import hkdf
|
||||
|
||||
import dbus
|
||||
import dbus.service
|
||||
import dbus.glib
|
||||
import gobject
|
||||
|
||||
bus_name = 'org.freedesktop.Secret.MockService'
|
||||
|
||||
class NotSupported(dbus.exceptions.DBusException):
|
||||
def __init__(self, msg):
|
||||
dbus.exceptions.DBusException.__init__(self, msg, name="org.freedesktop.DBus.Error.NotSupported")
|
||||
|
||||
unique_identifier = 0
|
||||
def next_identifier():
|
||||
global unique_identifier
|
||||
unique_identifier += 1
|
||||
return unique_identifier
|
||||
|
||||
class PlainAlgorithm():
|
||||
def negotiate(self, service, sender, param):
|
||||
session = SecretSession(service, sender, None)
|
||||
return (dbus.String("", variant_level=1), session)
|
||||
|
||||
class AesAlgorithm():
|
||||
def negotiate(self, service, sender, param):
|
||||
publi, privat = dh.generate_pair()
|
||||
peer = dh.bytes_to_number(param)
|
||||
ikm = dh.derive_key(privat, peer)
|
||||
key = hkdf.hkdf(ikm, 16)
|
||||
session = SecretSession(service, sender, key)
|
||||
return (dbus.ByteArray(dh.number_to_bytes(publi), variant_level=1), session)
|
||||
|
||||
class SecretSession(dbus.service.Object):
|
||||
def __init__(self, service, sender, key):
|
||||
self.sender = sender
|
||||
self.service = service
|
||||
self.key = key
|
||||
self.path = "/org/freedesktop/secrets/sessions/%d" % next_identifier()
|
||||
dbus.service.Object.__init__(self, service.bus_name, self.path)
|
||||
service.add_session(self)
|
||||
|
||||
@dbus.service.method('org.freedesktop.Secret.Session')
|
||||
def Close(self):
|
||||
self.remove_from_connection()
|
||||
self.service.remove_session(self)
|
||||
|
||||
class SecretService(dbus.service.Object):
|
||||
|
||||
algorithms = {
|
||||
'plain': PlainAlgorithm(),
|
||||
"dh-ietf1024-sha256-aes128-cbc-pkcs7": AesAlgorithm(),
|
||||
}
|
||||
|
||||
def __init__(self, name=bus_name):
|
||||
bus = bus = dbus.SessionBus()
|
||||
self.bus_name = dbus.service.BusName(name)
|
||||
dbus.service.Object.__init__(self, self.bus_name, '/org/freedesktop/secrets')
|
||||
self.sessions = { }
|
||||
|
||||
def on_name_owner_changed(owned, old_owner, new_owner):
|
||||
if not new_owner:
|
||||
for session in list(self.sessions.get(old_owner, [])):
|
||||
session.Close()
|
||||
|
||||
bus.add_signal_receiver(on_name_owner_changed,
|
||||
'NameOwnerChanged',
|
||||
'org.freedesktop.DBus')
|
||||
|
||||
def listen(self):
|
||||
loop = gobject.MainLoop()
|
||||
loop.run()
|
||||
|
||||
def add_session(self, session):
|
||||
if session.sender not in self.sessions:
|
||||
self.sessions[session.sender] = []
|
||||
self.sessions[session.sender].append(session)
|
||||
|
||||
def remove_session(self, session):
|
||||
self.sessions[session.sender].remove(session)
|
||||
|
||||
@dbus.service.method('org.freedesktop.Secret.Service', byte_arrays=True, sender_keyword='sender')
|
||||
def OpenSession(self, algorithm, param, sender=None):
|
||||
assert type(algorithm) == dbus.String
|
||||
assert type(param) == dbus.ByteArray
|
||||
|
||||
if algorithm not in self.algorithms:
|
||||
raise NotSupported("algorithm %s is not supported" % algorithm)
|
||||
|
||||
return self.algorithms[algorithm].negotiate(self, sender, param)
|
||||
|
||||
def parse_options(args):
|
||||
global bus_name
|
||||
try:
|
||||
opts, args = getopt.getopt(args, "name", ["name="])
|
||||
except getopt.GetoptError, err:
|
||||
print str(err)
|
||||
sys.exit(2)
|
||||
for o, a in opts:
|
||||
if o in ("--name"):
|
||||
bus_name = a
|
||||
else:
|
||||
assert False, "unhandled option"
|
||||
return args
|
||||
|
||||
parse_options(sys.argv[1:])
|
||||
@ -1,39 +0,0 @@
|
||||
/* GSecret - GLib wrapper for Secret Service
|
||||
*
|
||||
* Copyright 2011 Collabora Ltd.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Lesser General Public License as published
|
||||
* by the Free Software Foundation; either version 2 of the licence or (at
|
||||
* your option) any later version.
|
||||
*
|
||||
* See the included COPYING file for more information.
|
||||
*/
|
||||
|
||||
|
||||
#include "config.h"
|
||||
|
||||
#include "gsecret-item.h"
|
||||
#include "gsecret-service.h"
|
||||
|
||||
#include <glib.h>
|
||||
|
||||
static void
|
||||
test_initial (void)
|
||||
{
|
||||
GType type;
|
||||
|
||||
type = gsecret_service_get_type ();
|
||||
type += gsecret_item_get_type ();
|
||||
}
|
||||
|
||||
int
|
||||
main (int argc, char **argv)
|
||||
{
|
||||
g_test_init (&argc, &argv, NULL);
|
||||
g_set_prgname ("test-other");
|
||||
|
||||
g_test_add_func ("/initial", test_initial);
|
||||
|
||||
return g_test_run ();
|
||||
}
|
||||
89
library/tests/test-session.c
Normal file
89
library/tests/test-session.c
Normal file
@ -0,0 +1,89 @@
|
||||
/* GSecret - GLib wrapper for Secret Service
|
||||
*
|
||||
* Copyright 2011 Collabora Ltd.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Lesser General Public License as published
|
||||
* by the Free Software Foundation; either version 2 of the licence or (at
|
||||
* your option) any later version.
|
||||
*
|
||||
* See the included COPYING file for more information.
|
||||
*/
|
||||
|
||||
|
||||
#include "config.h"
|
||||
|
||||
#include "gsecret-item.h"
|
||||
#include "gsecret-service.h"
|
||||
#include "gsecret-private.h"
|
||||
|
||||
#include <glib.h>
|
||||
|
||||
#include <errno.h>
|
||||
#include <stdlib.h>
|
||||
|
||||
static gchar *MOCK_NAME = "org.mock.Service";
|
||||
|
||||
typedef struct {
|
||||
GPid pid;
|
||||
GDBusConnection *connection;
|
||||
GSecretService *service;
|
||||
} Test;
|
||||
|
||||
static void
|
||||
setup_normal (Test *test,
|
||||
gconstpointer unused)
|
||||
{
|
||||
GError *error = NULL;
|
||||
gchar *argv[] = {
|
||||
"python", "./mock-service-normal.py",
|
||||
"--name", MOCK_NAME,
|
||||
NULL
|
||||
};
|
||||
|
||||
g_spawn_async (SRCDIR, argv, NULL, G_SPAWN_SEARCH_PATH, NULL, NULL, &test->pid, &error);
|
||||
g_assert_no_error (error);
|
||||
|
||||
test->connection = g_bus_get_sync (G_BUS_TYPE_SESSION, NULL, &error);
|
||||
g_assert_no_error (error);
|
||||
|
||||
test->service = _gsecret_service_bare_instance (test->connection, MOCK_NAME);
|
||||
}
|
||||
|
||||
static void
|
||||
teardown (Test *test,
|
||||
gconstpointer unused)
|
||||
{
|
||||
g_clear_object (&test->service);
|
||||
g_clear_object (&test->connection);
|
||||
|
||||
g_assert (test->pid);
|
||||
if (kill (test->pid, SIGTERM) < 0)
|
||||
g_error ("kill() failed: %s", g_strerror (errno));
|
||||
g_spawn_close_pid (test->pid);
|
||||
}
|
||||
|
||||
static void
|
||||
test_ensure_sync (Test *test,
|
||||
gconstpointer unused)
|
||||
{
|
||||
GError *error = NULL;
|
||||
const gchar *path;
|
||||
|
||||
path = gsecret_service_ensure_session_sync (test->service, NULL, &error);
|
||||
g_assert_no_error (error);
|
||||
g_assert (path != NULL);
|
||||
g_printerr ("%s", path);
|
||||
}
|
||||
|
||||
int
|
||||
main (int argc, char **argv)
|
||||
{
|
||||
g_test_init (&argc, &argv, NULL);
|
||||
g_set_prgname ("test-session");
|
||||
g_type_init ();
|
||||
|
||||
g_test_add ("/session/ensure-sync", Test, NULL, setup_normal, test_ensure_sync, teardown);
|
||||
|
||||
return g_test_run ();
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user