foxhole/app/activitypub.py

#!/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 ap_id(self) -> RawObject:
        return self._ap_actor["id"]

    @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()