[or-cvs] r16902: {updater} More code for glider. (in updater/trunk/lib: glider sexp)
nickm at seul.org
nickm at seul.org
Sat Sep 13 14:53:59 UTC 2008
Author: nickm
Date: 2008-09-13 10:53:58 -0400 (Sat, 13 Sep 2008)
New Revision: 16902
Added:
updater/trunk/lib/glider/keys.py
updater/trunk/lib/glider/tests.py
Modified:
updater/trunk/lib/glider/
updater/trunk/lib/glider/formats.py
updater/trunk/lib/sexp/
Log:
More code for glider.
Property changes on: updater/trunk/lib/glider
___________________________________________________________________
Name: svn:ignore
+ *.pyc *.pyo
Modified: updater/trunk/lib/glider/formats.py
===================================================================
--- updater/trunk/lib/glider/formats.py 2008-09-13 10:21:27 UTC (rev 16901)
+++ updater/trunk/lib/glider/formats.py 2008-09-13 14:53:58 UTC (rev 16902)
@@ -1,28 +1,12 @@
-import OpenSSL.crypto
-
import sexp.access
import sexp.encode
import time
import re
-class UnknownMethod(Exception):
+class FormatException(Exception):
pass
-class PublicKey:
- def format(self):
- raise NotImplemented()
- def sign(self, data):
- # returns a list of method,signature tuples.
- raise NotImplemented()
- def checkSignature(self, method, data, signature):
- # returns True, False, or raises UnknownMethod.
- raise NotImplemented()
- def getKeyID(self):
- raise NotImplemented()
- def getRoles(self):
- raise NotImplemented()
-
class KeyDB:
def __init__(self):
self.keys = {}
@@ -31,18 +15,31 @@
def getKey(self, keyid):
return self.keys[keyid]
+_rolePathCache = {}
def rolePathMatches(rolePath, path):
"""
- >>> rolePath.matches("a/b/c/", "a/b/c/")
+ >>> rolePathMatches("a/b/c/", "a/b/c/")
True
- >>> rolePath.matches("**/c.*", "a/b/c.txt")
+ >>> rolePathMatches("**/c.*", "a/b/c.txt")
True
+ >>> rolePathMatches("**/c.*", "a/b/c.txt/foo")
+ False
+ >>> rolePathMatches("a/*/c", "a/b/c")
+ True
+ >>> rolePathMatches("a/*/c", "a/b/c.txt")
+ False
+ >>> rolePathMatches("a/*/c", "a/b/c.txt") #Check cache
+ False
"""
- rolePath = re.escape(rolePath).replace(r'\*\*', r'.*')
- rolePath = rolePath.replace(r'\*', r'[^/]*')
- rolePath += "$"
- return re.match(rolePath, path) != None
+ try:
+ regex = _rolePathCache[rolePath]
+ except KeyError:
+ rolePath = re.escape(rolePath).replace(r'\*\*', r'.*')
+ rolePath = rolePath.replace(r'\*', r'[^/]*')
+ rolePath += "$"
+ regex = _rolePathCache[rolePath] = re.compile(rolePath)
+ return regex.match(path) != None
def checkSignatures(signed, keyDB, role, path):
goodSigs = []
@@ -86,21 +83,41 @@
signed[2:] = oldsignatures
for method, sig in key.sign(s):
- signed.append(['signature', [['keyid', keyid], ['method', method]]
+ signed.append(['signature', [['keyid', keyid], ['method', method]],
sig])
def formatTime(t):
+ """
+ >>> formatTime(1221265172)
+ '2008-09-13 00:19:32'
+ """
return time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(t))
def parseTime(s):
return time.timegm(time.strptime(s, "%Y-%m-%d %H:%M:%S"))
+def _parseSchema(s, t=None):
+ sexpr = sexp.parse.parse(s)
+ schema = sexp.access.parseSchema(sexpr, t)
+ return schema
-TIME_SCHEMA = r"""/\{d}4-\d{2}-\d{2} \d{2}:\d{2}:\d{2}/"""
+SCHEMA_TABLE = { }
-ATTRS_SCHEMA = r"""(:anyof (_ *))"""
+PUBKEY_TEMPLATE = r"""
+ (=pubkey ((:unordered (=type .) (:anyof (. _)))) _)
+"""
-SIGNED_SCHEMA = r"""
+SCHEMA_TABLE['PUBKEY'] = _parseSchema(PUBKEY_TEMPLATE)
+
+TIME_TEMPLATE = r"""/\{d}4-\d{2}-\d{2} \d{2}:\d{2}:\d{2}/"""
+
+SCHEMA_TABLE['TIME'] = sexp.access.parseSchema(TIME_TEMPLATE)
+
+ATTRS_TEMPLATE = r"""(:anyof (_ *))"""
+
+SCHEMA_TABLE['ATTRS'] = _parseSchema(ATTRS_TEMPLATE)
+
+SIGNED_TEMPLATE = r"""
(=signed
_
(:someof
@@ -109,7 +126,9 @@
)
)"""
-KEYFILE_SCHEMA = r"""
+SIGNED_SCHEMA = _parseSchema(SIGNED_TEMPLATE, SCHEMA_TABLE)
+
+KEYFILE_TEMPLATE = r"""
(=keylist
(=ts .TIME)
(=keys
@@ -119,7 +138,9 @@
*
)"""
-MIRRORLIST_SCHEMA = r"""
+KEYFILE_SCHEMA = _parseSchema(KEYFILE_TEMPLATE, SCHEMA_TABLE)
+
+MIRRORLIST_TEMPLATE = r"""
(=mirrorlist
(=ts .TIME)
(=mirrors (:anyof
@@ -128,13 +149,17 @@
*)
"""
-TIMESTAMP_SCHEMA = r"""
+MIRRORLIST_SCHEMA = _parseSchema(MIRRORLIST_TEMPLATE, SCHEMA_TABLE)
+
+TIMESTAMP_TEMPLATE = r"""
(=ts
((:unordered (=at .TIME) (=m .TIME .) (=k .TIME .)
(:anyof (=b . . .TIME . .)) .ATTRS))
)"""
-BUNDLE_SCHEMA = r"""
+TIMESTAMP_SCHEMA = _parseSchema(TIMESTAMP_TEMPLATE, SCHEMA_TABLE)
+
+BUNDLE_TEMPLATE = r"""
(=bundle
(=at .TIME)
(=os .)
@@ -152,15 +177,19 @@
*
)"""
-PACKAGE_SCHEMA = r"""
+BUNDLE_SCHEMA = _parseSchema(BUNDLE_TEMPLATE, SCHEMA_TABLE)
+
+PACKAGE_TEMPLATE = r"""
(=package
- ((:unordred (=name .)
- (=version .)
- (=format . (.ATTRS))
- (=path .)
- (=ts .TIME)
- (=digest .)
- (:anyof (=shortdesc . .))
- (:anyof (=longdesc . .))
- .ATTRS)))
+ ((:unordered (=name .)
+ (=version .)
+ (=format . (.ATTRS))
+ (=path .)
+ (=ts .TIME)
+ (=digest .)
+ (:anyof (=shortdesc . .))
+ (:anyof (=longdesc . .))
+ .ATTRS)))
"""
+
+PACKAGE_SCHEMA = _parseSchema(PACKAGE_TEMPLATE, SCHEMA_TABLE)
Added: updater/trunk/lib/glider/keys.py
===================================================================
--- updater/trunk/lib/glider/keys.py (rev 0)
+++ updater/trunk/lib/glider/keys.py 2008-09-13 14:53:58 UTC (rev 16902)
@@ -0,0 +1,94 @@
+
+# These require PyCrypto.
+import Crypto.PublicKey.RSA
+import Crypto.Hash.SHA256
+
+import sexp.access
+import sexp.encode
+import sexp.parse
+
+import binascii
+import os
+
+class CryptoError(Exception):
+ pass
+
+class PubkeyFormatException(Exception):
+ pass
+
+class UnknownMethod(Exception):
+ pass
+
+class PublicKey:
+ def format(self):
+ raise NotImplemented()
+ def sign(self, data):
+ # returns a list of method,signature tuples.
+ raise NotImplemented()
+ def checkSignature(self, method, data, signature):
+ # returns True, False, or raises UnknownMethod.
+ raise NotImplemented()
+ def getKeyID(self):
+ raise NotImplemented()
+ def getRoles(self):
+ raise NotImplemented()
+
+def intToBinary(number):
+ h = hex(number)
+ assert h[:2] == '0x'
+ return binascii.a2b_hex(h[2:])
+
+def binaryToInt(binary):
+ return int(binascii.b2a_hex(binary), 16)
+
+def _pkcs1_padding(m, size):
+
+ # I'd rather use OAEP+, but apparently PyCrypto barely supports
+ # signature verification, and doesn't seem to support signature
+ # verification with nondeterministic padding. "argh."
+
+ s = [ "\x00\x01", "\xff"* (size-3-len(m)), "\x00", m ]
+ r = s.join()
+ return r
+
+class RSAKey(PublicKey):
+ def __init__(self, key):
+ self.key = key
+
+ @staticmethod
+ def generate(bits=2048):
+ key = Crypto.PublicKey.RSA.generate(bits=bits, randfunc=os.urandom)
+ return RSAKey(key)
+
+ @staticmethod
+ def fromSExpression(sexpr):
+ # sexpr must match PUBKEY_SCHEMA
+ typeattr = s_child(sexpr[1], "type")[1]
+ if typeattr[1] != "rsa":
+ return None
+ if len(sexpr[2]) != 2:
+ raise PubkeyFormatException("RSA keys must have an e,n pair")
+ e,n = sexpr[2]
+ key = Crypto.PublicKey.RSA.construct((binaryToInt(n), binaryToInt(e)))
+ return RSAKey(key)
+
+ def format(self):
+ n = intToBinary(self.key.n)
+ e = intToBinary(self.key.e)
+ return ("pubkey", [("type", "rsa")], (e, n))
+
+ def sign(self, sexpr):
+ d_obj = Crypto.Digest.SHA256.new()
+ sexpr.encode.hash_canonical(sexpr, d_obj)
+ m = _pkcs1_padding(d_obj.digest(), (self.key.size()+1) // 8)
+ return ("sha256-pkcs1", self.key.sign(m, "")[0])
+
+ def checkSignature(self, method, sexpr, sig):
+ if method != "sha256-pkcs1":
+ raise UnknownMethod("method")
+ d_obj = Crypto.Digest.SHA256.new()
+ sexpr.encode.hash_canonical(sexpr, d_obj)
+ m = _pkcs1_padding(d_obj.digest(), (self.key.size()+1) // 8)
+ return self.key.verify(sig, m)
+
+
Added: updater/trunk/lib/glider/tests.py
===================================================================
--- updater/trunk/lib/glider/tests.py (rev 0)
+++ updater/trunk/lib/glider/tests.py 2008-09-13 14:53:58 UTC (rev 16902)
@@ -0,0 +1,28 @@
+
+import unittest
+import doctest
+
+import glider.keys
+import glider.formats
+import glider.tests
+
+class EncodingTest(unittest.TestCase):
+ def testQuotedString(self):
+ self.assertEquals(1,1)
+
+def suite():
+ import sexp.tests
+ suite = unittest.TestSuite()
+
+ suite.addTest(doctest.DocTestSuite(glider.formats))
+ #suite.addTest(doctest.DocTestSuite(sexp.parse))
+
+ loader = unittest.TestLoader()
+ suite.addTest(loader.loadTestsFromModule(glider.tests))
+
+ return suite
+
+
+if __name__ == '__main__':
+
+ unittest.TextTestRunner(verbosity=1).run(suite())
Property changes on: updater/trunk/lib/sexp
___________________________________________________________________
Name: svn:ignore
+ *.pyc *.pyo
More information about the tor-commits
mailing list