feat/verify http signature

app/config.py

@@ -46,3 +46,6 @@ ID = f"{_SCHEME}://{DOMAIN}"
 BASE_URL = ID
 USERNAME = CONFIG.username
 KEY_PATH = (ROOT_DIR / "data" / "key.pem")
+
+USER_AGENT = "Fediverse Application/Foxhole-0.0.1"
+AP_CONTENT_TYPE = "application/activity+json"

app/httpsig.py

@@ -1,9 +1,11 @@
 #!/usr/bin/env python3
 import base64
 from typing import Literal, TypedDict, cast, Any
+from typing import Optional
 
 from Crypto.Hash import SHA256
 from Crypto.Signature import PKCS1_v1_5
+from Crypto.PublicKey import RSA
 
 class HttpSignature:
     """
@@ -11,9 +13,11 @@ class HttpSignature:
     """
 
     @classmethod
-    def calculation_digest(cls,
+    def calculation_digest(
+            cls,
             body : bytes,
-                               algorithm="sha-256")-> str :
+            algorithm : str ="sha-256"
+    )-> str :
         """
         Calculates the digest header value for a given HTTP body
         """
@@ -23,4 +27,51 @@ class HttpSignature:
             return "SHA-256=" + \
                 base64.b64encode(h.digest()).decode("utf-8")
         else:
-            raise ValueError(f"Not support algorithm {algorithm}")
+            raise ValueError(f"No support algorithm {algorithm}")
+
+    @classmethod
+    def verify_signature(
+            cls,
+            signature_string : str,
+            signature : bytes,
+            pubkey,
+    ) -> bool :
+        pubkey = RSA.importKey(pubkey)
+        signer = PKCS1_v1_5.new(pubkey)
+        digest = SHA256.new()
+        digest.update(signature_string.encode("utf-8"))
+        return signer.verify(digest, signature)
+
+    @classmethod
+    def parse_signature(cls, signature):
+        _detail = {}
+        for item in signature.split(","):
+            name, value = item.split("=", 1)
+            value = value.strip('"')
+            _detail[name.lower()] = value
+        signature_details = {
+            "headers": _detail["headers"].split(),
+            "signature": base64.b64decode(_detail["signature"]),
+            "algorithm": _detail["algorithm"],
+            "keyid": _detail["keyid"],
+        }
+        return signature_details
+
+    @classmethod
+    def build_signature_string(
+            cls,
+            method : str,
+            path : str,
+            signed_headers : dict,
+            body_digest : str,
+            headers,
+    ) -> str :
+        signed_string = []
+        for signed_header in signed_headers:
+            if signed_header == "(request-target)":
+                signed_string.append("(request-target): " + method.lower() + " " + path)
+            elif signed_header == "digest" and body_digest:
+                signed_string.append("digest: " + body_digest)
+            else:
+                signed_string.append(signed_header + ": " + headers[signed_header])
+        return "\n".join(signed_string)

app/main.py

@@ -1,6 +1,9 @@
 #!/usr/bin/env python3
+import logging
 import sys
 import fastapi
+import httpx
+import json
 
 from fastapi import FastAPI
 from fastapi import Depends
@@ -15,7 +18,7 @@ from loguru import logger
 
 from app import models
 from app.database import get_db_session
-from app.config import DEBUG
+from app.config import AP_CONTENT_TYPE, DEBUG, USER_AGENT
 from app.activitypub import ME
 from app.config import BASE_URL
 from app.config import DEBUG
@@ -24,6 +27,7 @@ from app.config import ID
 from app.config import USERNAME
 from app.database import AsyncSession
 from app.database import get_db_session
+from app.httpsig import HttpSignature
 
 def _check_0rtt_early_data(request: Request) -> None:
     """Disable TLS1.3 0-RTT requests for non-GET."""
@@ -47,8 +51,45 @@ async def inbox(
     db_session: AsyncSession = Depends(get_db_session),
 ) -> Response:
     logger.info(f"headers={request.headers}")
+
     payload = await request.json()
-    # logger.info(f"{payload=}")
+    logger.info(f"{payload=}")
+
+    parsec_signature = HttpSignature.parse_signature(
+        request.headers.get("signature"))
+
+    actor_url = payload["actor"]
+    async with httpx.AsyncClient() as client:
+        resp = await client.get(
+            actor_url,
+            headers={
+                "User-Agent": USER_AGENT,
+                "Accept": AP_CONTENT_TYPE,
+            },
+            follow_redirects=True,
+        )
+
+    try:
+        _actor = resp.json()
+        pubkey = _actor["publicKey"]["publicKeyPem"]
+    except json.JSONDecodeError:
+        raise ValueError
+
+    body = await request.body()
+    signture_string = HttpSignature.build_signature_string(
+        request.method,
+        request.url.path,
+        parsec_signature["headers"],
+        HttpSignature.calculation_digest(body),
+        request.headers,
+    )
+    is_verify = HttpSignature.verify_signature(
+        signture_string,
+        parsec_signature["signature"],
+        pubkey)
+    logger.info(signture_string)
+    logger.info(f"verify? {is_verify}")
+
     await new_incoming(db_session, payload)
     return Response(status_code=202)
 
@@ -70,9 +111,9 @@ async def new_incoming(
         ap_id=ap_id,
         ap_object=payload,
     )
-    db_session.add(incoming_activity)
-    await db_session.commit()
-    await db_session.refresh(incoming_activity)
+    # db_session.add(incoming_activity)
+    # await db_session.commit()
+    # await db_session.refresh(incoming_activity)
     return incoming_activity