ActivityPub

導航：首頁 > 程式語言、邏輯學> ActivityPub

概述

• 這是 Mastodon 等聯邦式社交網路的傳訊協定，構成 Fediverse。
• 類似 email 的傳輸協定。
• 傳輸使用 JSON。
• 要支援 Mastodon，需要使用 HTTP Signature。參見Message digest on http signatures check #4963

在 Python 支援 ActivityPub + HTTP Signature 的程式碼（未整理與加註解）

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

target_id = "https://dest.instance.social/users/peter"

recipient_url = target_id
recipient_inbox = target_id + "/inbox"



sender_url = "https://example.net/users/john"
sender_key = "https://example.net/users/john#main-key"

activity_id =  "https://example.net/users/john/follows/test"


# The following is to sign the HTTP request as defined in HTTP Signatures.
private_key_text = open('./lianlok/private.pem', 'rb').read() # load from file

private_key = crypto_serialization.load_pem_private_key(
    private_key_text,
    password=None,
    backend=crypto_default_backend()
)

current_date = datetime.datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')

recipient_parsed = urlparse(recipient_inbox)
recipient_host = recipient_parsed.netloc
recipient_path = recipient_parsed.path

# Now that the header is set up, we will construct the message
follow_request_message = {
    "@context": "https://www.w3.org/ns/activitystreams",
    "id": "https://example.net/users/john",
    "type": "Follow",
    "actor": sender_url,
    "object": recipient_url
}

# generating digest
request_message_json = json.dumps(follow_request_message)
digest = base64.b64encode(hashlib.sha256(request_message_json.__str__().encode('utf-8')).digest())

signature_text = b'(request-target): post %s\ndigest: SHA-256=%s\nhost: %s\ndate: %s' % (recipient_path.encode('utf-8'), digest, recipient_host.encode('utf-8'), current_date.encode('utf-8'))

raw_signature = private_key.sign(
    signature_text,
    padding.PKCS1v15(),
    hashes.SHA256()
)

signature_header = 'keyId="%s",algorithm="rsa-sha256",headers="(request-target) digest host date",signature="%s"' % (sender_key, base64.b64encode(raw_signature).decode('utf-8'))

headers = {
    'Date': current_date,
    'Content-Type': 'application/activity+json',
    'Host': recipient_host,
    'Digest': "SHA-256="+digest.decode('utf-8'),
    'Signature': signature_header
}




r = requests.post(recipient_inbox, headers=headers, json=follow_request_message)