1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
|
# License?
"""A U1DB implementation for using Object Stores as its persistence layer."""
import os
import string
import random
#import cStringIO
import hmac
import util
class Soledad(object):
PREFIX = os.environ['HOME'] + '/.config/leap/soledad'
SECRET_PATH = PREFIX + '/secret.gpg'
GNUPG_HOME = PREFIX + '/gnupg'
SECRET_LENGTH = 50
def __init__(self, user_email, gpghome=None):
self._user_email = user_email
if not os.path.isdir(self.PREFIX):
os.makedirs(self.PREFIX)
if not gpghome:
gpghome = self.GNUPG_HOME
self._gpg = util.GPGWrapper(gpghome=gpghome)
# load/generate OpenPGP keypair
if not self._has_openpgp_keypair():
self._gen_openpgp_keypair()
self._load_openpgp_keypair()
# load/generate secret
if not self._has_secret():
self._gen_secret()
self._load_secret()
#-------------------------------------------------------------------------
# Management of secret for symmetric encryption
#-------------------------------------------------------------------------
def _has_secret(self):
"""
Verify if secret already exists in a local encrypted file.
"""
if os.path.isfile(self.SECRET_PATH):
return True
return False
def _load_secret(self):
"""
Load secret from local encrypted file.
"""
try:
with open(self.SECRET_PATH) as f:
self._secret = str(self._gpg.decrypt(f.read()))
except IOError as e:
raise IOError('Failed to open secret file %s.' % self.SECRET_PATH)
def _gen_secret(self):
"""
Generate secret for symmetric encryption and store it in a local encrypted file.
"""
self._secret = ''.join(random.choice(string.ascii_uppercase + string.digits) for x in range(self.SECRET_LENGTH))
ciphertext = self._gpg.encrypt(self._secret, self._fingerprint, self._fingerprint)
f = open(self.SECRET_PATH, 'w')
f.write(str(ciphertext))
f.close()
#-------------------------------------------------------------------------
# Management of OpenPGP keypair
#-------------------------------------------------------------------------
def _has_openpgp_keypair(self):
"""
Verify if a keypair exists for this user.
"""
# TODO: verify if private key exists.
try:
self._gpg.find_key(self._user_email)
return True
except LookupError:
return False
def _gen_openpgp_keypair(self):
"""
Generate a keypair for this user.
"""
params = self._gpg.gen_key_input(
key_type='RSA',
key_length=4096,
name_real=self._user_email,
name_email=self._user_email,
name_comment='Generated by LEAP Soledad.')
self._gpg.gen_key(params)
def _load_openpgp_keypair(self):
"""
Load the fingerprint for this user's keypair.
"""
self._fingerprint = self._gpg.find_key(self._user_email)['fingerprint']
def publish_pubkey(self, keyserver):
"""
Publish OpenPGP public key to a keyserver.
"""
pass
#-------------------------------------------------------------------------
# Data encryption and decription
#-------------------------------------------------------------------------
def encrypt(self, data, sign=None, passphrase=None, symmetric=False):
"""
Encrypt data.
"""
return str(self._gpg.encrypt(data, self._fingerprint, sign=sign,
passphrase=passphrase, symmetric=symmetric))
def encrypt_symmetric(self, doc_id, data, sign=None):
"""
Symmetrically encrypt data using this user's secret.
"""
h = hmac.new(self._secret, doc_id).hexdigest()
return self.encrypt(data, sign=sign, passphrase=h, symmetric=True)
def decrypt(self, data, passphrase=None, symmetric=False):
"""
Decrypt data.
"""
return str(self._gpg.decrypt(data, passphrase=passphrase))
def decrypt_symmetric(self, doc_id, data):
"""
Symmetrically decrypt data using this user's secret.
"""
h = hmac.new(self._secret, doc_id).hexdigest()
return self.decrypt(data, passphrase=h)
#-------------------------------------------------------------------------
# Document storage, retrieval and sync
#-------------------------------------------------------------------------
def put(self, doc_id, data):
"""
Store a document.
"""
pass
def get(self, doc_id):
"""
Retrieve a document.
"""
pass
def sync(self):
"""
Synchronize with LEAP server.
"""
pass
__all__ = ['util']
|
