foxhole/app/boxes.py

#!/usr/bin/env python3
from app import models
from app.database import AsyncSession
from app.activitypub import ME
from loguru import logger
from uuid import uuid4
from app.config import MANUALLY_APPROVES_FOLLOWERS
from app.config import AP_CONTENT_TYPE
from app.config import BASE_URL
from app.config import USER_AGENT
from app.httpsig import auth

import app.activitypub as ap
import uuid
import httpx


def allocate_outbox_id() -> str:
    return str(uuid.uuid4())


def build_object_id(id) -> str:
    return f"{BASE_URL}/tail/{id}"

async def save_incoming(
        db_session: AsyncSession,
        payload: dict,
) -> models.IncomingActivity | None:
    ap_id: str
    if "@context" not in payload:
        logger.warning(f"invalid object: {payload}")
        return None

    if "id" in payload:
        ap_id = payload["id"]
    else:
        ap_id = str(uuid4())

    incoming_activity = models.IncomingActivity(
        ap_id=ap_id,
        ap_object=payload,
    )
    await process_incoming(db_session, payload)
    db_session.add(incoming_activity)
    await db_session.commit()
    await db_session.refresh(incoming_activity)
    return incoming_activity


async def process_incoming(
        db_session: AsyncSession,
        ap_object: dict,
) -> bool:

    if "Follow" == ap_object["type"]:
        await _handle_follow(db_session, ap_object)
    return True


async def _handle_follow(
        db_session: AsyncSession,
        ap_object: dict,
) -> None:
    if ME["id"] != ap_object["object"]:
        # await db_session.delete(ap_object)
        logger.warning("no match follow object!" + ap_object["object"])
        return

    if MANUALLY_APPROVES_FOLLOWERS:
        # TODO
        return

    await _send_accept(db_session, ap_object)


async def _send_accept(
        db_session: AsyncSession,
        ap_object: dict,
) -> None :

    reply_id = allocate_outbox_id()
    out = {
        "@context": ap.AS_CTX,
        "id": build_object_id(reply_id),
        "type": "Accept",
        "actor": ME["id"],
        "object": ap_object["id"],
    }

    async with httpx.AsyncClient() as client:
        resp = await client.post(
            ap_object["actor"] + "/inbox",
            headers={
                "User-Agent": USER_AGENT,
                "Content-Type": AP_CONTENT_TYPE,
            },
            json=out,
            auth=auth,
        )

    resp.raise_for_status()
    logger.info(resp)