Ticket #466: 2011-02-p2.diff

new file src/allmydata/scripts/admin.py

@@ +1 @@
+
+from twisted.python import usage
+
+class GenerateKeypairOptions(usage.Options):
+    def getSynopsis(self):
+        return "Usage: tahoe admin generate-keypair"
+
+    def getUsage(self, width=None):
+        t = usage.Options.getUsage(self, width)
+        t += """
+Generate an ECDSA192 public/private keypair, dumped to stdout as two lines of
+base32-encoded text.
+
+"""
+        return t
+
+def print_keypair(options):
+    from allmydata.util.keyutil import make_keypair
+    out = options.stdout
+    privkey_vs, pubkey_vs = make_keypair()
+    print >>out, "private:", privkey_vs
+    print >>out, "public:", pubkey_vs
+
+class DerivePubkeyOptions(usage.Options):
+    def parseArgs(self, privkey):
+        self.privkey = privkey
+
+    def getSynopsis(self):
+        return "Usage: tahoe admin derive-pubkey PRIVKEY"
+
+    def getUsage(self, width=None):
+        t = usage.Options.getUsage(self, width)
+        t += """
+Given a private (signing) key that was previously generated with
+generate-keypair, derive the public key and print it to stdout.
+
+"""
+        return t
+
+def derive_pubkey(options):
+    out = options.stdout
+    err = options.stderr
+    from allmydata.util import keyutil
+    privkey_vs = options.privkey
+    sk, pubkey_vs = keyutil.parse_privkey(privkey_vs)
+    print >>out, "private:", privkey_vs
+    print >>out, "public:", pubkey_vs
+    return 0
+
+class AdminCommand(usage.Options):
+    subCommands = [
+        ("generate-keypair", None, GenerateKeypairOptions,
+         "Generate a public/private keypair, write to stdout."),
+        ("derive-pubkey", None, DerivePubkeyOptions,
+         "Derive a public key from a private key."),
+        ]
+    def postOptions(self):
+        if not hasattr(self, 'subOptions'):
+            raise usage.UsageError("must specify a subcommand")
+    def getSynopsis(self):
+        return "Usage: tahoe admin SUBCOMMAND"
+    def getUsage(self, width=None):
+        t = usage.Options.getUsage(self, width)
+        t += """
+Please run e.g. 'tahoe admin generate-keypair --help' for more details on
+each subcommand.
+"""
+        return t
+
+subDispatch = {
+    "generate-keypair": print_keypair,
+    "derive-pubkey": derive_pubkey,
+    }
+
+def do_admin(options):
+    so = options.subOptions
+    so.stdout = options.stdout
+    so.stderr = options.stderr
+    f = subDispatch[options.subCommand]
+    return f(so)
+
+
+subCommands = [
+    ["admin", None, AdminCommand, "admin subcommands: use 'tahoe admin' for a list"],
+    ]
+
+dispatch = {
+    "admin": do_admin,
+    }

src/allmydata/scripts/runner.py

@@ -5 +5 @@
 from twisted.python import usage
 
 from allmydata.scripts.common import BaseOptions
-from allmydata.scripts import debug, create_node, startstop_node, cli, keygen, stats_gatherer
+from allmydata.scripts import debug, create_node, startstop_node, cli, keygen, stats_gatherer, admin
 from allmydata.util.encodingutil import quote_output, get_argv_encoding
 
 def GROUP(s):
@@ -21 +21 @@
                     +   create_node.subCommands
                     +   keygen.subCommands
                     +   stats_gatherer.subCommands
+                    +   admin.subCommands
                     + GROUP("Controlling a node")
                     +   startstop_node.subCommands
                     + GROUP("Debugging")
@@ -95 +96 @@
         rc = startstop_node.dispatch[command](so, stdout, stderr)
     elif command in debug.dispatch:
         rc = debug.dispatch[command](so)
+    elif command in admin.dispatch:
+        rc = admin.dispatch[command](so)
     elif command in cli.dispatch:
         rc = cli.dispatch[command](so)
     elif command in ac_dispatch:

src/allmydata/test/test_cli.py

@@ -7 +7 @@
 
 from mock import patch
 
-from allmydata.util import fileutil, hashutil, base32
+from allmydata.util import fileutil, hashutil, base32, keyutil
 from allmydata import uri
 from allmydata.immutable import upload
 from allmydata.dirnode import normalize
+from allmydata.util import ecdsa
 
 # Test that the scripts can be imported.
 from allmydata.scripts import create_node, debug, keygen, startstop_node, \
@@ -24 +25 @@
 from allmydata.scripts.common import DEFAULT_ALIAS, get_aliases, get_alias, \
      DefaultAliasMarker
 
-from allmydata.scripts import cli, debug, runner, backupdb
+from allmydata.scripts import cli, debug, runner, backupdb, admin
 from allmydata.test.common_util import StallMixin, ReallyEqualMixin
 from allmydata.test.no_network import GridTestMixin
 from twisted.internet import threads # CLI tests use deferToThread
@@ -1012 +1013 @@
 
         return d
 
+class Admin(unittest.TestCase):
+    def do_cli(self, *args, **kwargs):
+        argv = list(args)
+        stdin = kwargs.get("stdin", "")
+        stdout, stderr = StringIO(), StringIO()
+        d = threads.deferToThread(runner.runner, argv, run_by_human=False,
+                                  stdin=StringIO(stdin),
+                                  stdout=stdout, stderr=stderr)
+        def _done(res):
+            return stdout.getvalue(), stderr.getvalue()
+        d.addCallback(_done)
+        return d
+
+    def test_generate_keypair(self):
+        d = self.do_cli("admin", "generate-keypair")
+        def _done( (stdout, stderr) ):
+            lines = stdout.split("\n")
+            privkey_line = lines[0].strip()
+            pubkey_line = lines[1].strip()
+            sk_header = "private: priv-v0-"
+            vk_header = "public: pub-v0-"
+            self.failUnless(privkey_line.startswith(sk_header), privkey_line)
+            self.failUnless(pubkey_line.startswith(vk_header), pubkey_line)
+            privkey_b = base32.a2b(privkey_line[len(sk_header):])
+            pubkey_b = base32.a2b(pubkey_line[len(vk_header):])
+            sk = ecdsa.SigningKey.from_string(privkey_b)
+            vk = ecdsa.VerifyingKey.from_string(pubkey_b)
+            self.failUnlessEqual(sk.get_verifying_key().to_string(),
+                                 vk.to_string())
+        d.addCallback(_done)
+        return d
+
+    def test_derive_pubkey(self):
+        priv1,pub1 = keyutil.make_keypair()
+        d = self.do_cli("admin", "derive-pubkey", priv1)
+        def _done( (stdout, stderr) ):
+            lines = stdout.split("\n")
+            privkey_line = lines[0].strip()
+            pubkey_line = lines[1].strip()
+            sk_header = "private: priv-v0-"
+            vk_header = "public: pub-v0-"
+            self.failUnless(privkey_line.startswith(sk_header), privkey_line)
+            self.failUnless(pubkey_line.startswith(vk_header), pubkey_line)
+            pub2 = pubkey_line[len(vk_header):]
+            self.failUnlessEqual("pub-v0-"+pub2, pub1)
+        d.addCallback(_done)
+        return d
+
+
 class List(GridTestMixin, CLITestMixin, unittest.TestCase):
     def test_list(self):
         self.basedir = "cli/List/list"

new file src/allmydata/util/keyutil.py

@@ +1 @@
+from allmydata.util.ecdsa import SigningKey, VerifyingKey, NIST192p
+from allmydata.util import base32
+
+# in base32, the signing key is 39 chars long, and the verifying key is 77.
+# in base62, the signing key is 33 chars long, and the verifying key is 65.
+# in base64, the signing key is 32 chars long, and the verifying key is 64.
+#
+# We can't use base64 because we want to reserve punctuation and preserve
+# cut-and-pasteability. The base62 encoding is not significantly shorter than
+# the base32 form, and the minor usability improvement is not worth the
+# documentation/specification confusion of using a non-standard encoding. So
+# we stick with base32.
+
+def make_keypair():
+    privkey = SigningKey.generate(curve=NIST192p)
+    privkey_vs = "priv-v0-%s" % base32.b2a(privkey.to_string())
+    pubkey = privkey.get_verifying_key()
+    pubkey_vs = "pub-v0-%s" % base32.b2a(pubkey.to_string())
+    return privkey_vs, pubkey_vs
+
+def parse_privkey(privkey_vs):
+    if not privkey_vs.startswith("priv-v0-"):
+        raise ValueError("private key must look like 'priv-v0-...', not '%s'" % privkey_vs)
+    privkey_s = privkey_vs[len("priv-v0-"):]
+    sk = SigningKey.from_string(base32.a2b(privkey_s), curve=NIST192p)
+    pubkey = sk.get_verifying_key()
+    pubkey_vs = "pub-v0-%s" % base32.b2a(pubkey.to_string())
+    return sk, pubkey_vs
+
+def parse_pubkey(pubkey_vs):
+    if not pubkey_vs.startswith("pub-v0-"):
+        raise ValueError("public key must look like 'pub-v0-...', not '%s'" % pubkey_vs)
+    pubkey_s = pubkey_vs[len("pub-v0-"):]
+    vk = VerifyingKey.from_string(base32.a2b(pubkey_s), curve=NIST192p)
+    return vk