wip: first did implementation

2022-09-29 06:21:36 +02:00 · 2022-09-29 06:21:36 +02:00 · f1b4573294
parent c84b59e2cf
commit f1b4573294
5 changed files with 280 additions and 26 deletions
--- a/13
+++ b/13
@ -20,7 +20,7 @@
 set -eu

 DATE=$(date -u +%FT%TZ%Z)
-USAGE='[--help] [--version] show'
+USAGE='[--help] [--version] list'
 VERSION='0.0.1'

 dpgpid() {
@ -39,8 +39,8 @@ dpgpid() {
    -*)
      usage
      ;;
-    show)
-      show
+    list)
+      list
      ;;
    *)
      break
@ -54,8 +54,11 @@ help() {
  usage
 }

-show() {
-  gpg --list-keys --with-colons |awk -F: '$1 == "pub" {print "did:pgp:"$5}'
+list() {
+  gpg --list-secret-keys --with-colons |awk -F: '$1 == "sec" {print $5}' |while read key; do
+    id=$(keygen -t ipfs -g ${key} 2>/dev/null) && printf "did:ipid:${id}\n" ||
+      echo "ERROR: Unable to extract id from key ${key}"
+  done
 }

 usage() {
--- a/dpgpid.py
+++ b/dpgpid.py
@ -0,0 +1,245 @@
+#!/usr/bin/env python3
+# link: https://git.p2p.legal/aya/dpgpid/
+# desc: generate did:ipid keys from gpg
+
+# Copyleft 2022 Yann Autissier <aya@asycn.io>
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import argparse
+import configparser
+import gpg
+import ipfshttpclient
+import json
+import keygen
+import logging as log
+from SecureBytes import clearmem
+import os
+import re
+import struct
+import sys
+import time
+import warnings
+
+__version__='0.0.1'
+
+class dpgpid:
+    def __init__(self):
+        self.parser = argparse.ArgumentParser(description="""
+        Generate did:ipid keys from gpg.
+        It converts a gpg key to a Decentralized IDentifier.""")
+        self.parser.add_argument(
+            "-d",
+            "--debug",
+            action="store_true",
+            help="show debug informations",
+        )
+        self.parser.add_argument(
+            "-q",
+            "--quiet",
+            action="store_true",
+            help="show only errors",
+        )
+        self.parser.add_argument(
+            "-t",
+            "--type",
+            choices=['list','show','publish','base64','duniter','ipfs','jwk'],
+            default="list",
+            dest="type",
+            help="output text format, default: base58",
+        )
+        self.parser.add_argument(
+            "-v",
+            "--verbose",
+            action="store_true",
+            help="show more informations",
+        )
+        self.parser.add_argument(
+            "--version",
+            action="store_true",
+            help="show version and exit",
+        )
+        self.parser.add_argument(
+            'username',
+            nargs="?",
+        )
+        self.parser.add_argument(
+            'password',
+            nargs="?",
+        )
+
+    def _check_args(self, args):
+        log.debug("dpgpid._check_args(%s)" % args)
+        if False:
+            self.parser.error('dpgpid requires an input file or a username')
+
+    def _cleanup(self):
+        log.debug("dpgpid._cleanup()")
+        self.ipfs.close()
+
+    def _invalid_type(self):
+        log.debug("dpgpid._invalid_type()")
+        self.parser.error(f"type {self.type} is not valid.")
+
+    def _load_config(self):
+        log.debug("dpgpid._load_config()")
+        self.config = configparser.RawConfigParser()
+        config_dir = os.path.join(os.environ.get('XDG_CONFIG_HOME', os.path.expanduser('~/.config')), 'dpgpid')
+        log.debug("config_dir=%s" % config_dir)
+        self.config.read( [config_dir + '/dpgpid.conf'] )
+
+    def _run(self, argv):
+        args = self.parser.parse_args(argv)
+        vars(self).update(vars(args))
+
+        # display version
+        if args.version:
+            version()
+            sys.exit()
+
+        # define log format
+        log_format='%(asctime)s %(levelname)s: %(message)s'
+        log_datefmt='%Y/%m/%d %H:%M:%S'
+        if args.debug:
+            log_level='DEBUG'
+        elif args.quiet:
+            log_level='ERROR'
+        elif args.verbose:
+            log_level='INFO'
+        else:
+            log_level='WARNING'
+        log.basicConfig(format=log_format, datefmt=log_datefmt, level=log_level)
+
+        self._check_args(args)
+        self._load_config()
+        self.gpg = gpg.Context(armor=True, offline=True)
+        self.ipfs = ipfshttpclient.connect('/ip4/172.18.0.2/tcp/5001/',session=True)
+        method = getattr(self, f'do_{self.type}', self._invalid_type)
+        return method()
+
+    def did_from_key(self):
+        log.debug("dpgpid.did_from_key()")
+        self.did = json.dumps({
+"@context": "/ipfs/QmfS56jDfrXNaS6Xcsp3RJiXd2wyY7smeEAwyTAnL1RhEG",
+"id": "did:ipid:" + self.key_id,
+"created": self.key_created_at,
+"publicKey": [{
+  "id": "did:ipid:" + self.key_id,
+  "type": "GpgVerificationKey2020",
+  "expires": self.key_expires_at,
+  "publicKeyGpg": self.key_public_key,
+}, {
+  "id": "did:ipid:" + self.key_id,
+  "type": "JsonWebKey2020",
+  "expires": self.key_expires_at,
+  "publicKeyJwk": self.key_public_jwk,
+}],
+})
+        if self.key_updated_at:
+            self.did.update({"updated": self.key_updated_at,})
+        log.debug("dpgpid.did=%s" % self.did)
+
+    def do_list(self):
+        log.debug("dpgpid.do_list()")
+        gpgkeys = list(self.gpg.keylist(pattern=self.username, secret=True))
+        for gpgkey in gpgkeys:
+            self.key_from_gpg(gpgkey.fpr, self.password)
+            print("did:ipid:%s" % self.key_id)
+        self._cleanup()
+
+    def do_publish(self):
+        log.debug("dpgpid.do_publish()")
+        gpgkeys = list(self.gpg.keylist(pattern=self.username, secret=True))
+        if not gpgkeys:
+            log.warning(f"""Unable to find any key matching "{self.username}".""")
+            exit(1)
+        else:
+            gpgkey = gpgkeys[0]
+            log.info(f"""Found key id "{gpgkey.fpr}" matching "{self.username}".""")
+        self.key_from_gpg(gpgkey.fpr, self.password)
+        self.did_from_key()
+        self.did_cid = self.ipfs.add_json(self.did)
+        log.debug('dpgpid.did_cid=%s' % self.did_cid)
+        self.did_ipns = self.ipfs.name.publish(ipfs_path='/ipfs/' + self.did_cid, key=self.key_id)['Name']
+        log.debug('dpgpid.did_ipns=%s' % self.did_ipns)
+        self._cleanup()
+
+    def do_show(self):
+        log.debug("dpgpid.do_show()")
+        gpgkeys = list(self.gpg.keylist(pattern=self.username, secret=True))
+        if not gpgkeys:
+            log.warning(f"""Unable to find any key matching "{self.username}".""")
+            exit(1)
+        else:
+            gpgkey = gpgkeys[0]
+            log.info(f"""Found key id "{gpgkey.fpr}" matching "{self.username}".""")
+        self.key_from_gpg(gpgkey.fpr, self.password)
+        self.did_from_key()
+        print(self.did)
+        self._cleanup()
+ 
+    def key_from_gpg(self, username, password):
+        log.debug("dpgpid.key_from_gpg(%s, %s)" % (username, password))
+        try:
+            key = keygen.keygen()
+            key.gpg = gpg.Context(armor=True, offline=True)
+            key.gpg.set_passphrase_cb(key.gpg_passphrase_cb)
+            key.username = username
+            key.password = password
+            key.ed25519_from_gpg()
+            key.pem_pkcs8_from_ed25519()
+            key.protobuf_from_ed25519()
+            key.b58mh_from_protobuf()
+            key.jwk_from_ed25519()
+            self.key_created_at = str(key.pgpy.created)
+            self.key_expires_at = str(key.pgpy.expires_at)
+            self.key_fingerprint = key.gpg_secret_key.fpr
+            self.key_id = key.ed25519_public_b58mh
+            self.key_is_expired = key.gpg_secret_key.expired
+            self.key_is_revoked = key.gpg_secret_key.revoked
+            self.key_public_key = str(key.pgpy.pubkey)
+            self.key_public_jwk = key.ed25519_public_jwk
+            self.key_secret_jwk = key.ed25519_secret_jwk
+            self.key_secret_pem = key.ed25519_secret_pem_pkcs8
+            self.key_signers = key.pgpy.signers
+            self.key_uids = key.gpg_secret_key.uids
+            self.key_updated_at = key.gpg_secret_key.last_update
+            log.debug("dpgpid.key.created_at=%s" % self.key_created_at)
+            log.debug("dpgpid.key.expires_at=%s" % self.key_expires_at)
+            log.debug("dpgpid.key.fingerprint=%s" % self.key_fingerprint)
+            log.debug("dpgpid.key.id=%s" % self.key_id)
+            log.debug("dpgpid.key.is_expired=%s" % self.key_is_expired)
+            log.debug("dpgpid.key.is_revoked=%s" % self.key_is_revoked)
+            log.debug("dpgpid.key.public_key=%s" % self.key_public_key)
+            log.debug("dpgpid.key.public_jwk=%s" % self.key_public_jwk)
+            log.debug("dpgpid.key.secret_jwk=%s" % self.key_secret_jwk)
+            log.debug("dpgpid.key.signers=%s" % self.key_signers)
+            log.debug("dpgpid.key.uids=%s" % self.key_uids)
+            log.debug("dpgpid.key.updated_at=%s" % self.key_updated_at)
+        except Exception as e:
+            log.error(f'Unable to get key from gpg: {e}')
+            exit(2)
+
+def main(argv=None):
+    if argv is None:
+        argv = sys.argv[1:]
+
+    cli = dpgpid()
+    return cli._run(argv)
+
+def version(version=__version__):
+    print("%s v%s" % (sys.argv[0],version))
+
+if __name__ == "__main__":
+    sys.exit(main())
--- a/keygen.py
+++ b/keygen.py
@ -182,6 +182,9 @@ class keygen:
            clearmem(self.ipfs_privkey)
            log.debug("cleared: keygen.ipfs_privkey")
        if hasattr(self, 'jwk'):
+            if hasattr(self, 'ed25519_secret_jwk') and self.ed25519_secret_jwk:
+                clearmem(self.ed25519_secret_jwk)
+                log.debug("cleared: keygen.ed25519_secret_jwk")
            if hasattr(self.jwk, 'd') and self.jwk.d:
                clearmem(self.jwk.d)
                log.debug("cleared: keygen.jwk.d")
@ -330,7 +333,7 @@ class keygen:
        self._load_config()
        self.gpg = gpg.Context(armor=True, offline=True)
        self.gpg.set_passphrase_cb(self.gpg_passphrase_cb)
-        self.ed25519(args)
+        self.ed25519_from(args)
        method = getattr(self, f'do_{self.type}', self._invalid_type)
        return method()

@ -421,7 +424,7 @@ class keygen:
    def do_jwk(self):
        log.debug("keygen.do_jwk()")
        self.jwk_from_ed25519()
-        self._output(self.jwk.export_public(), self.jwk.export_private(), 'pub: ', 'sec: ')
+        self._output(self.ed25519_public_jwk, self.ed25519_secret_jwk, 'pub: ', 'sec: ')

    def duniterpy_from_credentials(self):
        log.debug("keygen.duniterpy_from_credentials()")
@ -591,8 +594,8 @@ class keygen:
            exit(2)
        log.debug("keygen.duniterpy.seed: %s" % self.duniterpy.seed)

-    def ed25519(self, args):
-        log.debug("keygen.ed25519(%s)" % args)
+    def ed25519_from(self, args):
+        log.debug("keygen.ed25519_from(%s)" % args)
        if args.gpg:
            self.ed25519_from_gpg()
        else:
@ -742,6 +745,8 @@ class keygen:
        log.debug("keygen.jwk_from_ed25519()")
        try:
            self.jwk = jwk.JWK.from_pyca(self.ed25519)
+            self.ed25519_public_jwk = self.jwk.export_public()
+            self.ed25519_secret_jwk = self.jwk.export_private()
        except Exception as e:
            log.error(f'Unable to get jwk from ed25519: {e}')
            self._cleanup()
@ -769,28 +774,28 @@ class keygen:
    def pgpy_from_gpg(self):
        log.debug("keygen.pgpy_from_gpg()")
        try:
-            self.gpg_seckeys = list(self.gpg.keylist(pattern=self.username, secret=True))
-            log.debug("keygen.gpg_seckeys=%s" % self.gpg_seckeys)
-            if not self.gpg_seckeys:
-                log.warning(f"""Unable to find any key matching username "{self.username}".""")
+            self.gpg_secret_keys = list(self.gpg.keylist(pattern=self.username, secret=True))
+            log.debug("keygen.gpg_secret_keys=%s" % self.gpg_secret_keys)
+            if not self.gpg_secret_keys:
+                log.warning(f"""Unable to find any key matching "{self.username}".""")
                self._cleanup()
                exit(1)
            else:
-                self.gpg_seckey = self.gpg_seckeys[0]
-                log.info(f"""Found key id "{self.gpg_seckey.fpr}" matching username "{self.username}".""")
-            log.debug("keygen.gpg_seckey.expired=%s" % self.gpg_seckey.expired)
-            log.debug("keygen.gpg_seckey.fpr=%s" % self.gpg_seckey.fpr)
-            log.debug("keygen.gpg_seckey.revoked=%s" % self.gpg_seckey.revoked)
-            log.debug("keygen.gpg_seckey.uids=%s" % self.gpg_seckey.uids)
-            log.debug("keygen.gpg_seckey.owner_trust=%s" % self.gpg_seckey.owner_trust)
-            log.debug("keygen.gpg_seckey.last_update=%s" % self.gpg_seckey.last_update)
+                self.gpg_secret_key = self.gpg_secret_keys[0]
+                log.info(f"""Found key id "{self.gpg_secret_key.fpr}" matching "{self.username}".""")
+            log.debug("keygen.gpg_secret_key.expired=%s" % self.gpg_secret_key.expired)
+            log.debug("keygen.gpg_secret_key.fpr=%s" % self.gpg_secret_key.fpr)
+            log.debug("keygen.gpg_secret_key.revoked=%s" % self.gpg_secret_key.revoked)
+            log.debug("keygen.gpg_secret_key.uids=%s" % self.gpg_secret_key.uids)
+            log.debug("keygen.gpg_secret_key.owner_trust=%s" % self.gpg_secret_key.owner_trust)
+            log.debug("keygen.gpg_secret_key.last_update=%s" % self.gpg_secret_key.last_update)
            if self.password:
                self.gpg.set_pinentry_mode(gpg.constants.PINENTRY_MODE_LOOPBACK)
-            self.pgp_public_armor = self.gpg.key_export(self.gpg_seckey.fpr)
-            self.pgp_secret_armor = self.gpg.key_export_secret(self.gpg_seckey.fpr)
+            self.pgp_public_armor = self.gpg.key_export(self.gpg_secret_key.fpr)
+            self.pgp_secret_armor = self.gpg.key_export_secret(self.gpg_secret_key.fpr)
            log.debug("keygen.pgp_secret_armor=%s" % self.pgp_secret_armor)
            if not self.pgp_secret_armor:
-                log.error(f"""Unable to export gpg secret key id "{self.gpg_seckey.fpr}" of user "{self.username}". Please check your password!""")
+                log.error(f"""Unable to export gpg secret key id "{self.gpg_secret_key.fpr}" of user "{self.username}". Please check your password!""")
                self._cleanup()
                exit(2)
            with warnings.catch_warnings():
--- a/requirements.txt
+++ b/requirements.txt
@ -1,6 +1,7 @@
 base58==2.1.1
 cryptography==3.4.8
 duniterpy==1.1.0
+ipfshttpclient==0.7.0
 jwcrypto==1.4.2
 pgpy==0.5.4
 pynentry==0.1.6
--- a/specs/keygen_spec.sh
+++ b/specs/keygen_spec.sh
@ -18,8 +18,8 @@ gpg() {
 }

 keygen() {
-  if [ -x ./keygen ]; then
-    GNUPGHOME="${SHELLSPEC_TMPBASE}" ./keygen "$@"
+  if [ -x ./keygen.py ]; then
+    GNUPGHOME="${SHELLSPEC_TMPBASE}" ./keygen.py "$@"
  elif [ -x ./bin/keygen ]; then
    GNUPGHOME="${SHELLSPEC_TMPBASE}" ./bin/keygen "$@"
  else