now also use RSA keys of pubgate

basic_ap.py

@@ -31,6 +31,7 @@ from Crypto.Signature import pkcs1_15
 from Crypto.PublicKey import RSA
 
 from httpsig import HTTPSigAuth
+from key import Key, get_key
 
 #
 import logging
@@ -61,73 +62,7 @@ def public_key():
         public_key = public_key.replace('\n','\\n') #public key shouldn't contain verbatim linebreaks in json
         return public_key
 
-public_key() #generate public_key on first launch
-
-# def build_signing_string(headers, header_values):
-#     return '\n'.join(map(lambda x: ': '.join([x.lower(), headers[x]]), header_values))
-
-# def messageContentDigest(messageBodyJsonStr: str) -> str:
-#     msg = messageBodyJsonStr.encode('utf-8')
-#     digestStr = SHA256.new(msg).digest()
-#     return base64.b64encode(digestStr).decode('utf-8')
-
-
-
-# def sign_header(private_key, keyID, host, messageBodyJsonStr):
-#     """
-#     Sign HTTP headers 
-#     """
-#     date = strftime("%a, %d %b %Y %H:%M:%S %Z", gmtime()) #RFC1123 Time format
-
-#     #Based on https://code.freedombone.net/bashrc/epicyon/src/main/httpsig.py
-#     keyfile = open(private_key,'rb').read()
-#     secret = RSA.import_key(keyfile) 
-
-#     bodyDigest = messageContentDigest(messageBodyJsonStr)
-#     contentLength = len(messageBodyJsonStr)
-#     print(contentLength)
-    
-#     path = '/inbox'
-
-#     headers = {
-#             '(request-target)': f'post {path}',
-#             'host': host,
-#             'date': date,
-#             'digest': f'SHA-256={bodyDigest}',
-#             'content-type': 'application/json',
-#             'content-length': str(contentLength)
-#         }
-
-
-#     signedHeaderKeys = headers.keys()
-#     signedHeaderText = ''
-
-#     for headerKey in signedHeaderKeys:
-#         signedHeaderText += f'{headerKey}: {headers[headerKey]}\n'
-
-#     signedHeaderText = signedHeaderText.strip()
-#     headerDigest = SHA256.new(signedHeaderText.encode('ascii'))
-
-#     # Sign the digest
-#     rawSignature = pkcs1_15.new(secret).sign(headerDigest)
-#     signature = base64.b64encode(rawSignature).decode('ascii')
-
-#     # Put it into a valid HTTP signature format
-#     signatureDict = {
-#         'keyId': keyID,
-#         'algorithm': 'rsa-sha256',
-#         'headers': ' '.join(signedHeaderKeys),
-#         'signature': signature
-#     }
-#     signatureHeader = ', '.join(
-#         [f'{k}="{v}"' for k, v in signatureDict.items()])
-    
-
-#     headers['signature'] = signatureHeader
-
-#     return headers
-
-
+key = get_key(f'{DOMAIN}/users/{USERNAME}') #generate public_key on first launch
 
 
 #Flask
@@ -159,34 +94,38 @@ def profile(actor):
     Return an Actor object 
     see templates/actor.json
     """
-    json = flask.render_template('actor.json', preferred_username=USERNAME, actor=actor, domain=DOMAIN, public_key=public_key()) # render our ActivityPub answer
+    key = get_key(f'{DOMAIN}/users/{actor}')
+    publicKeyPem = key.to_dict()['publicKeyPem']
+    publicKeyPem = publicKeyPem.replace('\n','\\n') #public key shouldn't contain verbatim linebreaks in json
+
+    json = flask.render_template('actor.json', preferred_username=USERNAME, actor=actor, domain=DOMAIN, public_key=publicKeyPem) # render our ActivityPub answer
+    
     return Response(response=json, status=200,  mimetype="application/json") # return that answer as a json object
 
 
 @app.route('/post/', methods=['POST','GET'])
 def post():
 
-    keyfile = open('private.pem','rb').read()
-    key = {'privkey': RSA.import_key(keyfile), 'keyId': DOMAIN+'/users/'+USERNAME+'#main-key'} 
+    #key = get_key(f'{DOMAIN}/users/{USERNAME}')
 
     headers = {"content-type": "application/activity+json",
-               "user-agent": f"test v:1"}
+               "user-agent": f"Basic AP v: 0.0"}
 
-    json_message=json.loads(flask.render_template('create.json',
-        domain=DOMAIN,public_key=public_key(),actor=USERNAME,host='DOMAIN'))
+    activity=json.loads(flask.render_template('create.json',
+        domain=DOMAIN, actor=USERNAME))
     
 
     http_sig = HTTPSigAuth(key, headers)
 
     url = 'https://post.lurk.org/inbox'
 
-    body = json.dumps(json_message)
+    body = json.dumps(activity)
 
     headers = http_sig.sign(url,body)
 
     #signed_headers = sign_header('private.pem', DOMAIN+'/users/'+USERNAME+'#main-key','https://post.lurk.org', json.dumps(json_message))
 
-    r = requests.post('https://post.lurk.org/inbox', json=body, headers=headers)
+    r = requests.post(url, json=body, headers=headers)
 
     html = f'Status code <pre>{r.status_code}</pre></br>Response Headers: <pre>{r.headers}</pre></br></br> json message: <pre>{r.request.body}</pre></br></br> Request headers: <pre>{r.request.headers}</pre>'
     html_headers = f'<html><head><style>pre {{overflow:auto;background-color:#eee;}}</style></head><body>{html}</body></html>'

httpsig.py

@@ -103,13 +103,13 @@ class HTTPSigAuth:
         sigheaders = headers.keys()
         sigstring = build_signing_string(headers, sigheaders)
 
-        signer = PKCS1_v1_5.new(self.key['privkey'])
+        signer = PKCS1_v1_5.new(self.key.privkey)
         digest = SHA256.new()
         digest.update(sigstring.encode("ascii"))
         sigdata = base64.b64encode(signer.sign(digest))
 
         sig = {
-            'keyId': self.key['keyId'],
+            'keyId': self.key.key_id(),
             'algorithm': 'rsa-sha256',
             'headers': ' '.join(sigheaders),
             'signature': sigdata.decode('ascii')

key.py

@@ -0,0 +1,74 @@
+import os
+import re
+import base64
+from Crypto.PublicKey import RSA
+from Crypto.Util import number
+from typing import Any
+from typing import Dict
+from typing import Optional
+
+KEY_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), '.')
+
+class Key(object):
+    DEFAULT_KEY_SIZE = 2048
+
+    def __init__(self, owner: str) -> None:
+        self.owner = owner
+        self.privkey_pem: Optional[str] = None
+        self.pubkey_pem: Optional[str] = None
+        self.privkey: Optional[Any] = None
+        self.pubkey: Optional[Any] = None
+
+    def load_pub(self, pubkey_pem: str) -> None:
+        self.pubkey_pem = pubkey_pem
+        self.pubkey = RSA.importKey(pubkey_pem)
+
+    def load(self, privkey_pem: str) -> None:
+        self.privkey_pem = privkey_pem
+        self.privkey = RSA.importKey(self.privkey_pem)
+        self.pubkey_pem = self.privkey.publickey().exportKey("PEM").decode("utf-8")
+
+    def new(self) -> None:
+        k = RSA.generate(self.DEFAULT_KEY_SIZE)
+        self.privkey_pem = k.exportKey("PEM").decode("utf-8")
+        self.pubkey_pem = k.publickey().exportKey("PEM").decode("utf-8")
+        self.privkey = k
+
+    def key_id(self) -> str:
+        return f"{self.owner}#main-key"
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "id": self.key_id(),
+            "owner": self.owner,
+            "publicKeyPem": self.pubkey_pem,
+        }
+
+    def to_magic_key(self) -> str:
+        mod = base64.urlsafe_b64encode(
+            number.long_to_bytes(self.privkey.n)  # type: ignore
+        ).decode("utf-8")
+        pubexp = base64.urlsafe_b64encode(
+            number.long_to_bytes(self.privkey.e)  # type: ignore
+        ).decode("utf-8")
+        return f"data:application/magic-public-key,RSA.{mod}.{pubexp}"
+
+
+def get_key(owner: str) -> Key:
+    """"Loads or generates an RSA key."""
+    k = Key(owner)
+    user = re.sub('[^\w\d]', "_", owner)
+    key_path = os.path.join(KEY_DIR, f"key_{user}.pem")
+    if os.path.isfile(key_path):
+        with open(key_path) as f:
+            privkey_pem = f.read()
+            k.load(privkey_pem)
+    else:
+        k.new()
+        with open(key_path, "w") as f:
+            f.write(k.privkey_pem)
+
+    return k
+
+
+