179 lines
4.8 KiB
Python
179 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
import httpx
|
|
import enum
|
|
|
|
from app import config
|
|
from app.key import get_pubkey_as_pem
|
|
from app.httpsig import auth
|
|
from loguru import logger
|
|
|
|
|
|
RawObject = dict
|
|
AS_CTX = "https://www.w3.org/ns/activitystreams"
|
|
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"},
|
|
"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",
|
|
# schema
|
|
"schema": "http://schema.org#",
|
|
"PropertyValue": "schema:PropertyValue",
|
|
"value": "schema:value",
|
|
# ostatus
|
|
"ostatus": "http://ostatus.org#",
|
|
"conversation": "ostatus:conversation",
|
|
},
|
|
]
|
|
|
|
_LOCAL_ACTOR_SUMMARY = config.CONFIG.summary
|
|
|
|
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.CONFIG.name,
|
|
"summary": _LOCAL_ACTOR_SUMMARY,
|
|
"endpoints": {
|
|
# For compat with servers expecting a sharedInbox...
|
|
"sharedInbox": config.BASE_URL
|
|
+ "/inbox",
|
|
},
|
|
"url": config.ID + "/", # the path is important for Mastodon compat
|
|
"manuallyApprovesFollowers": config.CONFIG.manually_approves_followers,
|
|
"attachment": [], # TODO media support
|
|
"icon": {
|
|
"mediaType": "image/png", # TODO media support
|
|
"type": "Image",
|
|
"url": config.CONFIG.icon_url,
|
|
},
|
|
"publicKey": {
|
|
"id": f"{config.ID}#main-key",
|
|
"owner": config.ID,
|
|
"publicKeyPem": get_pubkey_as_pem(config.KEY_PATH),
|
|
},
|
|
"tag": [] # TODO tag support
|
|
}
|
|
|
|
class BaseActor:
|
|
def __init__(self, ap_actor: RawObject) -> None:
|
|
if (ap_type := ap_actor.get("type")) not in ACTOR_TYPES:
|
|
raise ValueError(f"Unexpected actor type: {ap_type}")
|
|
|
|
self._ap_actor = ap_actor
|
|
self._ap_type : str = ap_type # type: ignore
|
|
|
|
@property
|
|
def ap_actor(self) -> RawObject:
|
|
return self._ap_actor
|
|
|
|
@property
|
|
def inbox_url(self) -> str:
|
|
return self.ap_actor["inbox"]
|
|
|
|
@property
|
|
def ap_type(self) -> str:
|
|
return self._ap_type
|
|
|
|
@property
|
|
def share_inbox_url(self) -> str:
|
|
return self.ap_actor.get("endpoints", {}).get("sharedInbox") \
|
|
or self.inbox_url
|
|
|
|
|
|
class VisibilityEnum(str, enum.Enum):
|
|
PUBLIC = "public"
|
|
UNLISTED = "unlisted"
|
|
FOLLOWERS_ONLY = "followers-only"
|
|
DIRECT = "direct"
|
|
|
|
|
|
def handle_visibility(
|
|
ap_object: dict
|
|
) -> VisibilityEnum:
|
|
to = ap_object.get("to", [])
|
|
cc = ap_object.get("cc", [])
|
|
if AS_PUBLIC in to:
|
|
return VisibilityEnum.PUBLIC
|
|
elif AS_PUBLIC in cc:
|
|
return VisibilityEnum.UNLISTED
|
|
else:
|
|
return VisibilityEnum.DIRECT
|
|
|
|
|
|
def wrap_ap_object(ap_object: dict) -> dict:
|
|
if ap_object["type"] in ["Note"]:
|
|
|
|
if "@context" in ap_object:
|
|
del ap_object["@context"]
|
|
|
|
return {
|
|
"@context": AS_EXTENDED_CTX,
|
|
"actor": config.ID,
|
|
"to": ap_object.get("to", []),
|
|
"cc": ap_object.get("cc", []),
|
|
"id": ap_object["id"] + "/activity",
|
|
"object": ap_object,
|
|
"published": ap_object["published"],
|
|
"type": "Create",
|
|
}
|
|
|
|
return ap_object
|
|
|
|
|
|
async def post(
|
|
url: str,
|
|
payload : dict,
|
|
) -> httpx.Response :
|
|
logger.info(f"send post, {payload=}")
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.post(
|
|
url,
|
|
headers={
|
|
"User-Agent": config.USER_AGENT,
|
|
"Content-Type": config.AP_CONTENT_TYPE,
|
|
},
|
|
json=payload,
|
|
auth=auth,
|
|
)
|
|
|
|
resp.raise_for_status()
|
|
return resp
|
|
|
|
|
|
async def fetch(
|
|
url: str,
|
|
) -> dict:
|
|
logger.info(f"fetch {url}")
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(
|
|
url,
|
|
headers={
|
|
"User-Agent": config.USER_AGENT,
|
|
"Accept": config.AP_CONTENT_TYPE,
|
|
},
|
|
)
|
|
|
|
return resp.json()
|