1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
| # 因為要加密，所以要 import 一堆東西。
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 # 分析 url
import base64 #用 base64 轉碼用
import datetime #時間日期
import requests #送 http request
import hashlib #產生雜湊用
import json # 處理 json
import os.path # 處理路徑
default_sender = "alice" # 寄件者名稱
sender_domain = "activitypub_test.ddns.net" #寄件者域名
receiver_name = "bob" #收件人名稱
recipient_domain = "example.social" #收件者域名
'''產生一些連結'''
recipient_url = f"https://{recipient_domain}/users/{receiver_name}" #對方的代表網址
recipient_inbox = f"https://{recipient_domain}/users/{receiver_name}/inbox" #對方的寄件匣
sender_url = f"https://{sender_domain}/users/{default_sender}" # 發信者用戶網址
sender_key = f"https://{sender_domain}/users/{default_sender}#main-key" # 發信者的 key
# 這個舉動（發送 follow request）的 identified（獨一）的識別網址，這裏使用 test，實際可以用亂數
activity_id = f"https://{sender_domain}/users/{default_sender}/follows/test"
'''簽署 http request'''
home_folder = os.path.expanduser("~") # 家目錄
private_key_path = os.path.join(home_folder, "my_folder/my_keys/private.pem")
private_key_path = f"/etc/letsencrypt/live/{sender_domain}/privkey.pem" # 私鑰路徑
private_key_text = open(private_key_path, 'rb').read() # load from file
# 將 private_key_text 的文字存成 private_key
private_key = crypto_serialization.load_pem_private_key(
private_key_text,
password=None,
backend=crypto_default_backend()
)
# 產生格式如 Mon, 21 Nov 2022 14:47:28 GMT 的現在時間（UTC 時區）
current_date = datetime.datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')
recipient_parsed = urlparse(recipient_inbox) #分析收件方的地址
recipient_path = recipient_parsed.path # '/users/bob/inbox'
# following 的訊息內容 (dict)
follow_request_message = {
"@context": "https://www.w3.org/ns/activitystreams",
"id": activity_id,
"type": "Follow",
"actor": sender_url, # 收件者地址
"object": recipient_url # 寄件者地址
}
'''製作digest摘要資訊'''
# 將follow_request_message 轉 json
follow_request_json = json.dumps(follow_request_message)
#產生上述 json 的 hash 值，並用 base64 編碼，存成 digest（摘要）
digest = base64.b64encode(hashlib.sha256(follow_request_json.__str__().encode('utf-8')).digest())
# 建立簽署用文字
# 格式為：
#(request-target): post [recipient_path]
#digest: [digest]
#host: [recipient_domain]
#date: [current_date]
# 方框[]內的值用 utf-8 編碼
signature_text = b'(request-target): post %s\ndigest: SHA-256=%s\nhost: %s\ndate: %s' % \
(recipient_path.encode('utf-8'),
digest,
recipient_domain.encode('utf-8'),
current_date.encode('utf-8'))
# 用私鑰產生 signature_text 簽署得出的值
raw_signature = private_key.sign(
signature_text,
padding.PKCS1v15(),
hashes.SHA256()
)
# 將簽署值轉成 base64
raw_signature_in_base64 = base64.b64encode(raw_signature).decode('utf-8')
print(signature_text)
# 提供寄件者公鑰 sender_key 和轉成 base64 的簽署值，存到 signature_header
# headers="(request-target) digest host date 和上面提到的簽署用文字格式相似
signature_header = f'keyId="{sender_key}",algorithm="rsa-sha256",headers="(request-target) digest host date",signature="{raw_signature_in_base64}"'
# 設定 request 的 http 表頭
headers = {
'Date': current_date, # 日期
'Content-Type': 'application/activity+json',
'Host': recipient_domain, # 收件者域名
'Digest': "SHA-256="+digest.decode('utf-8'), # digest 用 UTF-8 解碼
'Signature': signature_header
}
# 以 POST 方法送出 request，回應存在 r
r = requests.post(recipient_inbox,
headers=headers, # 表頭
json=follow_request_message # 內容
)
print(r, r.content) # 顯示回應內容
|