Add direct example

timmot · timmot · commit 9279f22cbefd · 2024-12-17T21:44:37.000+11:00
diff --git a/example/README.md b/example/README.md
@@ -0,0 +1,23 @@
+# ActivityPub Example
+
+Here is a less wordy example which will let you see some results immediately.
+
+1. Set up a server using Let's Encrypt to get your HTTPS certificates.
+2. Turn off any web servers you may have used, e.g. nginx/apache/caddy
+3. Generate a public/private key pair for your user
+    ```sh
+    openssl genrsa -out private.pem 2048
+    openssl rsa -in private.pem -outform PEM -pubout -out public.pem
+    ```
+4. Set up Python environment
+    ```sh
+    python3 -m venv env
+    source env/bin/activate
+    pip install -r requirements.txt
+    ```
+5. Enter your own details in config.py
+6. Run the server in one shell `python server.py`
+    a. This may need to run as root as it binds port 443
+    b. Confirm by searching your user on Mastodon, e.g. `@user@host.com` and note what response you receive
+7. Run `python follow_user.py` and `python follow_user.py --unfollow`
+    a. The server must be running for Mastodon to send an accept message.
diff --git a/example/config.py b/example/config.py
@@ -0,0 +1,8 @@
+# Limit yourself to one test user otherwise every username will be valid
+# and Mastodon will cache these until you provide it with a Delete action
+HOSTNAME = "example.com"
+USER = "johnny"
+
+# Ideally use a user account you own on a working ActivityPub instance
+TEST_HOSTNAME = "example.com" # mastodon.social
+TEST_USER = "zampano"
diff --git a/example/follow_user.py b/example/follow_user.py
@@ -0,0 +1,220 @@
+from cryptography.hazmat.backends import default_backend as crypto_default_backend
+from cryptography.hazmat.primitives import serialization as crypto_serialization
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import padding
+
+from urllib.parse import urlparse
+import base64
+import datetime
+import requests
+import json
+import hashlib
+import sys
+from config import HOSTNAME, USER, TEST_HOSTNAME, TEST_USER
+
+
+OWNED_URI = f"https://{HOSTNAME}"
+OWNED_USER = USER
+
+sender_url = f"{OWNED_URI}/users/{OWNED_USER}"
+sender_key = f"{OWNED_URI}/users/{OWNED_USER}#main-key"
+# NOTE: This id should be unique for each action
+activity_id = f"{OWNED_URI}/users/{OWNED_USER}/follows/test"
+
+TEST_URI = f"https://{TEST_HOSTNAME}"
+
+
+def get_resource_from_webfinger(uri, resource):
+    print("Getting recipient url from .well-known/webfinger")
+    webfinger = requests.get(
+        f"{uri}/.well-known/webfinger?resource={resource}",
+        headers={"Accept": "application/jrd+json, application/json"},
+    )
+    return json.loads(webfinger.content)
+
+
+def get_canonical_from_webfinger(uri, resource):
+    webfinger_json = get_resource_from_webfinger(uri, resource)
+
+    if "links" not in webfinger_json:
+        print("No links in webfinger")
+        return None
+
+    self_uri = [
+        link["href"] for link in webfinger_json["links"] if link["rel"] == "self"
+    ]
+    if len(self_uri) == 0:
+        print("No self uri")
+        return None
+
+    return self_uri[0]
+
+
+def get_inbox_from_canonical_user(uri, resource):
+    recipient_url = get_canonical_from_webfinger(uri, resource)
+
+    user = requests.get(
+        recipient_url, headers={"Accept": "application/activity+json, application/json"}
+    )
+    user_json = json.loads(user.content)
+
+    if "inbox" not in user_json:
+        print("No inbox")
+        return None
+
+    return user_json["inbox"]
+
+
+def sign(text):
+    # The following is to sign the HTTP request as defined in HTTP Signatures.
+    private_key_text = open("private.pem", "rb").read()  # load from file
+
+    private_key = crypto_serialization.load_pem_private_key(
+        private_key_text, password=None, backend=crypto_default_backend()
+    )
+    return private_key.sign(text, padding.PKCS1v15(), hashes.SHA256())
+
+
+def follow():
+    recipient_url = get_canonical_from_webfinger(
+        TEST_URI, f"acct:{TEST_USER}@{TEST_HOSTNAME}"
+    )
+    recipient_inbox = get_inbox_from_canonical_user(
+        TEST_URI, f"acct:{TEST_USER}@{TEST_HOSTNAME}"
+    )
+
+    print(f"Sending follow request from {sender_url} to {recipient_inbox}")
+
+    recipient_parsed = urlparse(recipient_inbox)
+    recipient_host = recipient_parsed.netloc
+    recipient_path = recipient_parsed.path
+
+    follow_request_message = {
+        "@context": "https://www.w3.org/ns/activitystreams",
+        "id": activity_id,
+        "type": "Follow",
+        "actor": sender_url,
+        "object": recipient_url,
+    }
+
+    digest = base64.b64encode(
+        hashlib.sha256(json.dumps(follow_request_message).encode("utf-8")).digest()
+    )
+
+    current_date = datetime.datetime.now(datetime.timezone.utc).strftime(
+        "%a, %d %b %Y %H:%M:%S GMT"
+    )
+
+    # signature_text = generate_signing_text
+    signature_text: bytes = b"(request-target): post " + recipient_path.encode("utf-8")
+    signature_text += b"\nhost: " + recipient_host.encode("utf-8")
+    signature_text += b"\ndate: " + current_date.encode("utf-8")
+    signature_text += b"\ndigest: SHA-256=" + digest
+
+    raw_signature = sign(signature_text)
+
+    signature_text_b64 = base64.b64encode(raw_signature).decode("utf-8")
+    signature_header = (
+        'keyId="'
+        + sender_key
+        + '",algorithm="rsa-sha256",headers="(request-target) host date digest",signature="'
+        + signature_text_b64
+        + '"'
+    )
+
+    headers = {
+        "Date": current_date,
+        "Content-Type": "application/activity+json",
+        "Host": recipient_host,
+        "Digest": "SHA-256=" + digest.decode("utf-8"),
+        "Signature": signature_header,
+    }
+    print(headers)
+
+    # Now that the header is set up, we will construct the message
+    r = requests.post(recipient_inbox, headers=headers, json=follow_request_message)
+    print(r)
+    print(r.content)
+
+
+def unfollow():
+    recipient_url = get_canonical_from_webfinger(
+        TEST_URI, f"acct:{TEST_USER}@{TEST_HOSTNAME}"
+    )
+    recipient_inbox = get_inbox_from_canonical_user(
+        TEST_URI, f"acct:{TEST_USER}@{TEST_HOSTNAME}"
+    )
+
+    print(f"Sending unfollow request from {sender_url} to {recipient_inbox}")
+
+    recipient_parsed = urlparse(recipient_inbox)
+    recipient_host = recipient_parsed.netloc
+    recipient_path = recipient_parsed.path
+
+    follow_request_message = {
+        "id": activity_id,
+        "type": "Follow",
+        "actor": sender_url,
+        "object": recipient_url,
+    }
+
+    unfollow_request_message = {
+        "@context": "https://www.w3.org/ns/activitystreams",
+        "id": f"{activity_id}/undo",
+        "type": "Undo",
+        "actor": sender_url,
+        "object": follow_request_message,
+    }
+
+    digest = base64.b64encode(
+        hashlib.sha256(json.dumps(unfollow_request_message).encode("utf-8")).digest()
+    )
+
+    current_date = datetime.datetime.now(datetime.timezone.utc).strftime(
+        "%a, %d %b %Y %H:%M:%S GMT"
+    )
+
+    # signature_text = generate_signing_text
+    signature_text: bytes = b"(request-target): post " + recipient_path.encode("utf-8")
+    signature_text += b"\nhost: " + recipient_host.encode("utf-8")
+    signature_text += b"\ndate: " + current_date.encode("utf-8")
+    signature_text += b"\ndigest: SHA-256=" + digest
+
+    raw_signature = sign(signature_text)
+
+    signature_text_b64 = base64.b64encode(raw_signature).decode("utf-8")
+    signature_header = (
+        'keyId="'
+        + sender_key
+        + '",algorithm="rsa-sha256",headers="(request-target) host date digest",signature="'
+        + signature_text_b64
+        + '"'
+    )
+
+    headers = {
+        "Date": current_date,
+        "Content-Type": "application/activity+json",
+        "Host": recipient_host,
+        "Digest": "SHA-256=" + digest.decode("utf-8"),
+        "Signature": signature_header,
+    }
+    print(headers)
+
+    # Now that the header is set up, we will construct the message
+    r = requests.post(recipient_inbox, headers=headers, json=unfollow_request_message)
+    print(r)
+    print(r.content)
+
+
+if __name__ == "__main__":
+    command = "follow"
+    if len(sys.argv) > 1:
+        if sys.argv[1] == "--unfollow":
+            command = "unfollow"
+
+    if command == "follow":
+        follow()
+    elif command == "unfollow":
+        unfollow()
+    else:
+        print("Command not handled")
diff --git a/example/requirements.txt b/example/requirements.txt
@@ -0,0 +1,3 @@
+cryptography==38.0.1
+requests==2.28.1
+flask==3.1.0
diff --git a/example/server.py b/example/server.py
@@ -0,0 +1,102 @@
+from flask import Flask, request, make_response, abort, Response
+import ssl
+from config import HOSTNAME, USER
+
+URI = f"https://{HOSTNAME}"
+
+context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+context.load_cert_chain(
+    f"/etc/letsencrypt/live/{HOSTNAME}/fullchain.pem",
+    f"/etc/letsencrypt/live/{HOSTNAME}/privkey.pem",
+)
+
+app = Flask(__name__)
+
+
+@app.route("/users/<username>")
+def user(username):
+    print(f"Received GET request for /users/{username}")
+    print(request.headers)
+
+    if username != USER:
+        abort(404)
+
+    public_key = open("public.pem", "r").read()
+
+    response = make_response(
+        {
+            "@context": [
+                "https://www.w3.org/ns/activitystreams",
+                "https://w3id.org/security/v1",
+            ],
+            "id": f"{URI}/users/{USER}",
+            "inbox": f"{URI}/users/{USER}/inbox",
+            "outbox": f"{URI}/users/{USER}/outbox",
+            "type": "Person",
+            "name": USER.title(),
+            "preferredUsername": USER,
+            "publicKey": {
+                "id": f"{URI}/users/{USER}#main-key",
+                "owner": f"{URI}/users/{USER}",
+                "publicKeyPem": public_key,
+            },
+        }
+    )
+
+    # Servers may discard the result if you do not set the appropriate content type
+    response.headers["Content-Type"] = "application/activity+json"
+
+    return response
+
+
+@app.route("/users/<username>/inbox", methods=["POST"])
+def user_inbox(username):
+    print(f"Received POST request for /users/{username}/inbox")
+    print(request.headers)
+    print(request.data)
+
+    if username != USER:
+        abort(404)
+
+    # Accept any message sent to our inbox while testing
+    return Response("", status=202)
+
+
+@app.route("/.well-known/webfinger")
+def webfinger():
+    resource = request.args.get("resource")
+
+    if resource != f"acct:{USER}@{HOSTNAME}":
+        abort(404)
+
+    response = make_response(
+        {
+            "subject": f"acct:{USER}@{HOSTNAME}",
+            "links": [
+                {
+                    "rel": "self",
+                    "type": "application/activity+json",
+                    "href": f"{URI}/users/{USER}",
+                }
+            ],
+        }
+    )
+
+    # Servers may discard the result if you do not set the appropriate content type
+    response.headers["Content-Type"] = "application/jrd+json"
+
+    return response
+
+
+@app.route("/", defaults={"path": ""})
+@app.route("/<path:path>")
+def catch_all(path):
+    print(path)
+    print(request.headers)
+    print(request.data)
+
+    return ""
+
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=443, ssl_context=context)

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+cryptography==38.0.1`
	`2`	`+requests==2.28.1`
	`3`	`+flask==3.1.0`