[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]

[or-cvs] r16917: {updater} More glider hacks: implement public keys. (in updater/trunk/lib: glider sexp)



Author: nickm
Date: 2008-09-16 12:07:40 -0400 (Tue, 16 Sep 2008)
New Revision: 16917

Modified:
   updater/trunk/lib/glider/formats.py
   updater/trunk/lib/glider/keys.py
   updater/trunk/lib/glider/tests.py
   updater/trunk/lib/sexp/access.py
Log:
More glider hacks: implement public keys.

Modified: updater/trunk/lib/glider/formats.py
===================================================================
--- updater/trunk/lib/glider/formats.py	2008-09-16 10:26:15 UTC (rev 16916)
+++ updater/trunk/lib/glider/formats.py	2008-09-16 16:07:40 UTC (rev 16917)
@@ -35,6 +35,7 @@
     try:
         regex = _rolePathCache[rolePath]
     except KeyError:
+        rolePath = re.sub(r'/+', '/', rolePath)
         rolePath = re.escape(rolePath).replace(r'\*\*', r'.*')
         rolePath = rolePath.replace(r'\*', r'[^/]*')
         rolePath += "$"

Modified: updater/trunk/lib/glider/keys.py
===================================================================
--- updater/trunk/lib/glider/keys.py	2008-09-16 10:26:15 UTC (rev 16916)
+++ updater/trunk/lib/glider/keys.py	2008-09-16 16:07:40 UTC (rev 16917)
@@ -33,13 +33,33 @@
     def getRoles(self):
         raise NotImplemented()
 
-def intToBinary(number):
-    h = hex(number)
-    assert h[:2] == '0x'
-    return binascii.a2b_hex(h[2:])
+if hex(1L).upper() == "0X1L":
+    def intToBinary(number):
+        """Convert an int or long into a big-endian series of bytes.
+        """
+        # This "convert-to-hex, then use binascii" approach may look silly,
+        # but it's over 10x faster than the Crypto.Util.number approach.
+        h = hex(long(number))
+        h = h[2:-1]
+        if len(h)%2:
+            h = "0"+h
+        return binascii.a2b_hex(h)
+elif hex(1L).upper() == "0X1":
+    def intToBinary(number):
+        h = hex(long(number))
+        h = h[2:]
+        if len(h)%2:
+            h = "0"+h
+        return binascii.a2b_hex(h)
+else:
+    import Crypto.Util.number
+    intToBinary = Crypto.Util.number.long_to_bytes
+    assert None
 
 def binaryToInt(binary):
-    return int(binascii.b2a_hex(binary), 16)
+   """Convert a big-endian series of bytes into a long.
+   """
+   return long(binascii.b2a_hex(binary), 16)
 
 def _pkcs1_padding(m, size):
 
@@ -48,12 +68,39 @@
     # verification with nondeterministic padding.  "argh."
 
     s = [ "\x00\x01", "\xff"* (size-3-len(m)), "\x00", m ]
-    r = s.join()
+    r = "".join(s)
     return r
 
+def _xor(a,b):
+    if a:
+        return not b
+    else:
+        return b
+
 class RSAKey(PublicKey):
+    """
+    >>> k = RSAKey.generate(bits=512)
+    >>> sexpr = k.format()
+    >>> sexpr[:2]
+    ('pubkey', [('type', 'rsa')])
+    >>> k1 = RSAKey.fromSExpression(sexpr)
+    >>> k1.key.e == k.key.e
+    True
+    >>> k1.key.n == k.key.n
+    True
+    >>> k.getKeyID() == k1.getKeyID()
+    True
+    >>> s = ['tag1', ['foobar'], [['foop', 'bar']], 'baz']
+    >>> method, sig = k.sign(sexpr=s)
+    >>> k.checkSignature(method, sig, sexpr=s)
+    True
+    >>> s2 = [ s ]
+    >>> k.checkSignature(method, sig, sexpr=s2)
+    False
+    """
     def __init__(self, key):
         self.key = key
+        self.keyid = None
 
     @staticmethod
     def generate(bits=2048):
@@ -63,8 +110,8 @@
     @staticmethod
     def fromSExpression(sexpr):
         # sexpr must match PUBKEY_SCHEMA
-        typeattr = s_child(sexpr[1], "type")[1]
-        if typeattr[1] != "rsa":
+        typeattr = sexp.access.s_attr(sexpr[1], "type")
+        if typeattr != "rsa":
             return None
         if len(sexpr[2]) != 2:
             raise PubkeyFormatException("RSA keys must have an e,n pair")
@@ -77,18 +124,36 @@
         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 getKeyID(self):
+        if self.keyid == None:
+            n = intToBinary(self.key.n)
+            e = intToBinary(self.key.e)
+            keyval = (e,n)
+            d_obj = Crypto.Hash.SHA256.new()
+            sexp.encode.hash_canonical(keyval, d_obj)
+            self.keyid = ("rsa", d_obj.digest())
+        return self.keyid
 
-    def checkSignature(self, method, sexpr, sig):
+    def sign(self, sexpr=None, digest=None):
+        assert _xor(sexpr == None, digest == None)
+        if digest == None:
+            d_obj = Crypto.Hash.SHA256.new()
+            sexp.encode.hash_canonical(sexpr, d_obj)
+            digest = d_obj.digest()
+        m = _pkcs1_padding(digest, (self.key.size()+1) // 8)
+        sig = intToBinary(self.key.sign(m, "")[0])
+        return ("sha256-pkcs1", sig)
+
+    def checkSignature(self, method, sig, sexpr=None, digest=None):
+        assert _xor(sexpr == None, digest == None)
         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)
+        if digest == None:
+            d_obj = Crypto.Hash.SHA256.new()
+            sexp.encode.hash_canonical(sexpr, d_obj)
+            digest = d_obj.digest()
+        sig = binaryToInt(sig)
+        m = _pkcs1_padding(digest, (self.key.size()+1) // 8)
+        return self.key.verify(m, (sig,))
 
 

Modified: updater/trunk/lib/glider/tests.py
===================================================================
--- updater/trunk/lib/glider/tests.py	2008-09-16 10:26:15 UTC (rev 16916)
+++ updater/trunk/lib/glider/tests.py	2008-09-16 16:07:40 UTC (rev 16917)
@@ -15,7 +15,7 @@
     suite = unittest.TestSuite()
 
     suite.addTest(doctest.DocTestSuite(glider.formats))
-    #suite.addTest(doctest.DocTestSuite(sexp.parse))
+    suite.addTest(doctest.DocTestSuite(glider.keys))
 
     loader = unittest.TestLoader()
     suite.addTest(loader.loadTestsFromModule(glider.tests))

Modified: updater/trunk/lib/sexp/access.py
===================================================================
--- updater/trunk/lib/sexp/access.py	2008-09-16 10:26:15 UTC (rev 16916)
+++ updater/trunk/lib/sexp/access.py	2008-09-16 16:07:40 UTC (rev 16917)
@@ -36,6 +36,16 @@
             return child
     return None
 
+def s_attr(s, tag):
+    """Returns the second element of the child of 's' whose tag is 'tag'.
+       This is helpful for extracting a (key val) element.  Returns None
+       if there is no such element.
+    """
+    ch = s_child(s,tag)
+    if ch == None or len(ch) < 2:
+        return None
+    return ch[1]
+
 def s_children(s, tag):
     """Returns a generator yielding all children of 's' whose tag is 'tag'.