COSCUP-ap-demo/docs/show.org

Show

• 名正言顺
  • Nodeinfo
  • Actor
• WebFinger
• HTTP Signature

名正言顺

Nodeinfo

from flask import Flask, Response, jsonify
from config import BASE_URL #http://coscup.localhost

app = Flask(__name__)
@app.route("/.well-known/nodeinfo")
def well_known_nodeinfo() -> Response:
    """Return nodeinfo path."""
    return jsonify(
        {
            "links": [
                {
                    "rel": "http://nodeinfo.diaspora.software/ns/schema/2.0",
                    "href": f"{BASE_URL}/nodeinfo/2.0",
                }
            ]
        }
    )

@app.get("/nodeinfo/2.0")
def nodeinfo() -> Response:
    """Return nodeinfo."""
    return jsonify(
        {
            "version": "2.0",
            "software": {
                "name": "COSCUP-demo",
                "version": "0.0.1",
            },
            "protocols": ["activitypub"],
            "services": {"inbound": [], "outbound": []},
            "usage": {"users": {"total": 1}},
            "openRegistrations": False,
            "metadata": {},
        },
    )

GET http://coscup.localhost/.well-known/nodeinfo

GET http://coscup.localhost/nodeinfo/2.0

Actor

AS_CTX = "https://www.w3.org/ns/activitystreams"
AS_PUBLIC = "https://www.w3.org/ns/activitystreams#Public"

AS_EXTENDED_CTX = [
    "https://www.w3.org/ns/activitystreams",
    "https://w3id.org/security/v1",
    {
        # AS ext
        "Hashtag": "as:Hashtag",
        "sensitive": "as:sensitive",
        "manuallyApprovesFollowers": "as:manuallyApprovesFollowers",
        "alsoKnownAs": {"@id": "as:alsoKnownAs", "@type": "@id"},
        "movedTo": {"@id": "as:movedTo", "@type": "@id"},
        # toot
        "toot": "http://joinmastodon.org/ns#",
        "featured": {"@id": "toot:featured", "@type": "@id"},
        "Emoji": "toot:Emoji",
        "blurhash": "toot:blurhash",
        "votersCount": "toot:votersCount",
    },
]

ME = {
    "@context": AS_EXTENDED_CTX,
    "type": "Person",
    "id": config.ID,
    "following": config.BASE_URL + "/following",
    "followers": config.BASE_URL + "/followers",
    "featured": config.BASE_URL + "/featured",
    "inbox": config.BASE_URL + "/inbox",
    "outbox": config.BASE_URL + "/outbox",
    "preferredUsername": config.USERNAME,
    "name": config.NICKNAME,
    "summary": config.ACTOR_SUMMARY,
    "endpoints": {
        "sharedInbox": config.BASE_URL + "/inbox",
    },
    "url": config.ID + "/",  # Important for Mastodon
    "manuallyApprovesFollowers": False,
    "attachment": [],
    "icon": {
        "mediaType": "image/png",
        "type": "Image",
        "url": config.AVATAR_URL,
    },
    "publicKey": {
        "id": f"{config.ID}#main-key",
        "owner": config.ID,
        "publicKeyPem": get_pubkey_as_pem(config.KEY_PATH),
    },
    "tag": [],
}

@app.route(f"/user/{config.USERNAME}") # /user/show
def locate_user() -> Response:
    """Return user ActivityPub response."""
    resp = jsonify(ME)
    resp.headers["Content-Type"] = "application/activity+json"

    return resp

GET http://coscup.localhost/user/show

WebFinger

from flask import Flask, abort, request, Response, jsonify
from config import USERNAME, DOMAIN, ID
#USERNAME = "show"
#DOMAIN = "coscup.localhost"
#ID = f"http://{DOMAIN}/meow/{USERNAME}"

@app.route("/.well-known/webfinger")
def wellknown_webfinger() -> Response:
    """Exposes servers WebFinger data."""
    resource = request.args.get("resource")

    if resource not in [f"acct:{USERNAME}@{DOMAIN}", config.ID]:
        abort(404)

    resp = jsonify(
        {
            "subject": f"acct:{config.USERNAME}@{config.DOMAIN}",
            "aliases": [config.ID],
            "links": [
                {
                    "rel": "http://webfinger.net/rel/profile-page",
                    "type": "text/html",
                    "href": config.ID,
                },
                {
                    "rel": "self",
                    "type": "application/activity+json",
                    "href": config.ID,
                },
            ],
        },
    )
    resp.headers["Access-Control-Allow-Origin"]  = "*"
    resp.headers["Content-Type"] = "application/jrd+json; charset=utf-8"

    return resp

GET http://coscup.localhost/.well-known/webfinger?resource=acct:show@coscup.localhost

HTTP Signature

other.localhost –Follow-> coscup.localhost

# request.data
{
   "@context": "https://www.w3.org/ns/activitystreams",
   "id": "http://other.localhost/9ef8847d72434cff82254f36ff88e710",
   "type": "Follow",
   "actor": "http://other.localhost",
   "object": "http://coscup.localhost/meow/show"
}

# request.header
{
   "host":"other.localhost",
   "accept":"*/*",
   "accept-encoding":"gzip, deflate, br",
   "connection":"keep-alive",
   "user-agent":"something/fediverse+app",
   "content-type":"application/activity+json",
   "content-length":"206",
   "date":"Tue, 11 Jul 2023 04:21:11 GMT",
   "digest":"SHA-256=WzNniTyRBcZpgK7a6zYJBgsdKIQmFEPPfysh/ZzKb2o=",
   "signature":"keyId=\"http://other.localhost#main-key\",algorithm=\"rsa-sha256\",headers=\"(request-target) user-agent host date digest content-type\",signature=\"irioDQuhYstSvafpl6DW4d31wDPRiTv7MZGyBo3j4kkc2TrfOweH3WRMMnoaWwl4LAI2WYLoKefeQpOg7Rm7ZEffsoLOzZvgdWJBm8lnOEgieyy5l2Vq1mlcS2PRJCisYGdzAwFOBkcHk0WKAZXvs1ieRV34NHfM8JF+DjrCBTZ/U9LyxULBwC6tPQTh9tflCOwXZOzXUq17C+2Uzsr8h4tDHjbmrG7OAcvYiPeOUKaP+InoE6j9ViHllhidNCPL0y8b1c7c72ruN48kF42OfyfUeiuCcuLwdp8eYBlTdG/ZsT2YXyKruwim3tTD1TtyW4Vfll+F/4/1RfWHsc9LrQ==\""
}

def parse_signature(
        signature: str
) -> dict:
    """Parse signature string in headers."""
    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": (detail["signature"]),
        "algorithm": detail["algorithm"],
        "keyid": detail["keyid"],
    }
    return signature_details

parse_signature(headers["signature"])

 {"algorithm": "rsa-sha256",
  "headers": ["(request-target)", "user-agent", "host", "date", "digest", "content-type"],
  "keyid": "http://other.localhost#main-key",
  "signature": "irioDQuhYstSvafpl6DW4d31wDPRiTv7MZGyBo3j4kkc2TrfOweH3WRMMnoaWwl4LAI2WYLoKefeQpOg7Rm7ZEffsoLOzZvgdWJBm8lnOEgieyy5l2Vq1mlcS2PRJCisYGdzAwFOBkcHk0WKAZXvs1ieRV34NHfM8JF+DjrCBTZ/U9LyxULBwC6tPQTh9tflCOwXZOzXUq17C+2Uzsr8h4tDHjbmrG7OAcvYiPeOUKaP+InoE6j9ViHllhidNCPL0y8b1c7c72ruN48kF42OfyfUeiuCcuLwdp8eYBlTdG/ZsT2YXyKruwim3tTD1TtyW4Vfll+F/4/1RfWHsc9LrQ=="}

import base64
from Crypto.Hash import SHA256

def calculate_digest(
        payload: bytes,
) -> str:
    """Calculate digest for given HTTP payload."""
    payload_digest = SHA256.new()
    payload_digest.update(body)
    return "SHA-256=" + \
        base64.b64encode(body_digest.digest()).decode("utf-8")


def build_signature_string(
        method: str,
        path: str,
        sign_headers: list,
        payload_digest: str | None,
        headers,
) -> str :
    sign_str = []
    for sign_header in sign_headers:
        if sign_header == "(request-target)":
            sign_str.append("(request-target): " + method.lower() + ' ' + path)
        elif sign_header == "digest" and payload_digest:
            sign_str.append("digest: " + payload_digest)
        else:
            sign_str.append(sign_header + ": " + headers[sign_header])
    return "\n".join(sign_str)


build_signature_string(
    request.method  # "post"
    request.path    # "/inbox"
    parse_signature(headers["signature"])["headers"], # ["(request-target)", "user-agent", "host", "date", "digest", "content-type"]
    calculate_digest(request.data),
    headers,
)

(request-target): post /inbox user-agent: something/fediverse+app host: other.localhost date: Tue, 11 Jul 2023 04:21:11 GMT digest: SHA-256=WzNniTyRBcZpgK7a6zYJBgsdKIQmFEPPfysh/ZzKb2o= content-type: application/activity+json