Cleanup context and LD sig

main
Thomas Sileo 2022-06-28 09:58:33 +02:00
parent e01a9ddbe4
commit 489ed6cbe0
5 changed files with 91 additions and 74 deletions

View File

@ -20,6 +20,30 @@ AS_PUBLIC = "https://www.w3.org/ns/activitystreams#Public"
ACTOR_TYPES = ["Application", "Group", "Organization", "Person", "Service"]
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"},
# toot
"toot": "http://joinmastodon.org/ns#",
"featured": {"@id": "toot:featured", "@type": "@id"},
"Emoji": "toot:Emoji",
"blurhash": "toot:blurhash",
# schema
"schema": "http://schema.org#",
"PropertyValue": "schema:PropertyValue",
"value": "schema:value",
# ostatus
"ostatus": "http://ostatus.org#",
"conversation": "ostatus:conversation",
},
]
class ObjectIsGoneError(Exception):
pass
@ -41,45 +65,8 @@ class VisibilityEnum(str, enum.Enum):
}[key]
MICROBLOGPUB = {
"@context": [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/security/v1",
{
"Hashtag": "as:Hashtag",
"PropertyValue": "schema:PropertyValue",
"manuallyApprovesFollowers": "as:manuallyApprovesFollowers",
"ostatus": "http://ostatus.org#",
"schema": "http://schema.org",
"sensitive": "as:sensitive",
"toot": "http://joinmastodon.org/ns#",
"totalItems": "as:totalItems",
"value": "schema:value",
"Emoji": "toot:Emoji",
},
]
}
DEFAULT_CTX = COLLECTION_CTX = [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/security/v1",
{
# AS ext
"Hashtag": "as:Hashtag",
"sensitive": "as:sensitive",
"manuallyApprovesFollowers": "as:manuallyApprovesFollowers",
# toot
"toot": "http://joinmastodon.org/ns#",
# "featured": "toot:featured",
# schema
"schema": "http://schema.org#",
"PropertyValue": "schema:PropertyValue",
"value": "schema:value",
},
]
ME = {
"@context": DEFAULT_CTX,
"@context": AS_EXTENDED_CTX,
"type": "Person",
"id": config.ID,
"following": config.BASE_URL + "/following",
@ -235,7 +222,7 @@ def get_actor_id(activity: RawObject) -> str:
def wrap_object(activity: RawObject) -> RawObject:
return {
"@context": AS_CTX,
"@context": AS_EXTENDED_CTX,
"actor": config.ID,
"to": activity.get("to", []),
"cc": activity.get("cc", []),

View File

@ -273,7 +273,7 @@ def send_create(
raise ValueError(f"Unhandled visibility {visibility}")
note = {
"@context": ap.AS_CTX,
"@context": ap.AS_EXTENDED_CTX,
"type": "Note",
"id": outbox_object_id(note_id),
"attributedTo": ID,

View File

@ -2,29 +2,31 @@ import base64
import hashlib
import typing
from datetime import datetime
from functools import lru_cache
from typing import Any
import pyld # type: ignore
from Crypto.Hash import SHA256
from Crypto.Signature import PKCS1_v1_5
from pyld import jsonld # type: ignore
from app import activitypub as ap
if typing.TYPE_CHECKING:
from app.key import Key
_LOADER = jsonld.requests_document_loader()
requests_loader = pyld.documentloader.requests.requests_document_loader()
@lru_cache(256)
def _caching_document_loader(url: str) -> Any:
return _LOADER(url)
def _loader(url, options={}):
# See https://github.com/digitalbazaar/pyld/issues/133
options["headers"]["Accept"] = "application/ld+json"
return requests_loader(url, options)
jsonld.set_document_loader(_caching_document_loader)
pyld.jsonld.set_document_loader(_loader)
def _options_hash(doc):
def _options_hash(doc: ap.RawObject) -> str:
doc = dict(doc["signature"])
for k in ["type", "id", "signatureValue"]:
if k in doc:
@ -38,7 +40,7 @@ def _options_hash(doc):
return h.hexdigest()
def _doc_hash(doc):
def _doc_hash(doc: ap.RawObject) -> str:
doc = dict(doc)
if "signature" in doc:
del doc["signature"]
@ -50,7 +52,7 @@ def _doc_hash(doc):
return h.hexdigest()
def verify_signature(doc, key: "Key"):
def verify_signature(doc: ap.RawObject, key: "Key") -> bool:
to_be_signed = _options_hash(doc) + _doc_hash(doc)
signature = doc["signature"]["signatureValue"]
signer = PKCS1_v1_5.new(key.pubkey or key.privkey) # type: ignore
@ -59,7 +61,7 @@ def verify_signature(doc, key: "Key"):
return signer.verify(digest, base64.b64decode(signature)) # type: ignore
def generate_signature(doc, key: "Key"):
def generate_signature(doc: ap.RawObject, key: "Key") -> None:
options = {
"type": "RsaSignature2017",
"creator": doc["actor"] + "#main-key",

View File

@ -133,25 +133,6 @@ async def add_security_headers(request: Request, call_next):
return response
DEFAULT_CTX = COLLECTION_CTX = [
"https://www.w3.org/ns/activitystreams",
"https://w3id.org/security/v1",
{
# AS ext
"Hashtag": "as:Hashtag",
"sensitive": "as:sensitive",
"manuallyApprovesFollowers": "as:manuallyApprovesFollowers",
# toot
"toot": "http://joinmastodon.org/ns#",
# "featured": "toot:featured",
# schema
"schema": "http://schema.org#",
"PropertyValue": "schema:PropertyValue",
"value": "schema:value",
},
]
class ActivityPubResponse(JSONResponse):
media_type = "application/activity+json"
@ -372,7 +353,7 @@ def outbox(
)
return ActivityPubResponse(
{
"@context": DEFAULT_CTX,
"@context": ap.AS_EXTENDED_CTX,
"id": f"{ID}/outbox",
"type": "OrderedCollection",
"totalItems": len(outbox_objects),
@ -402,7 +383,7 @@ def featured(
)
return ActivityPubResponse(
{
"@context": DEFAULT_CTX,
"@context": ap.AS_EXTENDED_CTX,
"id": f"{ID}/featured",
"type": "OrderedCollection",
"totalItems": len(outbox_objects),
@ -512,7 +493,7 @@ def tag_by_name(
# if is_activitypub_requested(request):
return ActivityPubResponse(
{
"@context": ap.AS_CTX,
"@context": ap.AS_CTX, # XXX: extended ctx?
"id": BASE_URL + f"/t/{tag}",
"type": "OrderedCollection",
"totalItems": 0,
@ -528,7 +509,7 @@ def emoji_by_name(name: str) -> ActivityPubResponse:
except KeyError:
raise HTTPException(status_code=404)
return ActivityPubResponse({"@context": ap.AS_CTX, **emoji})
return ActivityPubResponse({"@context": ap.AS_EXTENDED_CTX, **emoji})
@app.post("/inbox")

View File

@ -0,0 +1,47 @@
from copy import deepcopy
import pytest
from app import activitypub as ap
from app import ldsig
from app.key import Key
from tests import factories
_SAMPLE_CREATE = {
"type": "Create",
"actor": "https://microblog.pub",
"object": {
"type": "Note",
"sensitive": False,
"cc": ["https://microblog.pub/followers"],
"to": ["https://www.w3.org/ns/activitystreams#Public"],
"content": "<p>Hello world!</p>",
"tag": [],
"attributedTo": "https://microblog.pub",
"published": "2018-05-21T15:51:59Z",
"id": "https://microblog.pub/outbox/988179f13c78b3a7/activity",
"url": "https://microblog.pub/note/988179f13c78b3a7",
},
"@context": ap.AS_EXTENDED_CTX,
"published": "2018-05-21T15:51:59Z",
"to": ["https://www.w3.org/ns/activitystreams#Public"],
"cc": ["https://microblog.pub/followers"],
"id": "https://microblog.pub/outbox/988179f13c78b3a7",
}
@pytest.mark.skip(reason="Working but slow")
def test_linked_data_sig():
privkey, pubkey = factories.generate_key()
ra = factories.RemoteActorFactory(
base_url="https://microblog.pub",
username="dev",
public_key=pubkey,
)
k = Key(ra.ap_id, f"{ra.ap_id}#main-key")
k.load(privkey)
doc = deepcopy(_SAMPLE_CREATE)
ldsig.generate_signature(doc, k)
assert ldsig.verify_signature(doc, k)