An exploration into an ActivityPub Minimal Core.
from typing import Any

import httpx
import uvicorn
from fastapi import APIRouter, Request
from fastapi.exceptions import HTTPException
from fastapi.responses import JSONResponse
from jsonschema import Draft202012Validator

from activitypub_mincore.support.validation import MINCORE_REGISTRY, schema_retriever

router = APIRouter()

_actor: dict = {}


@router.get("/{path:path}")
async def get__actor(request: Request):
    if str(request.url) == _actor["id"]:
        return JSONResponse(get_local_actor())
    raise HTTPException(404, "Not found")


def initialize_actor(server: uvicorn.Server):
    global _actor
    url_prefix = f"http://{server.config.host}:{server.config.port}"
    _actor = {
        "id": f"{url_prefix}/actor",
        "type": "Service",
        "inbox": f"{url_prefix}/inbox",
        "outbox": f"{url_prefix}/outbox",
    }
    server.config.app.include_router(router)


def get_local_actor() -> dict[str, Any]:
    return _actor


ACTOR_VALIDATOR = Draft202012Validator(
    schema_retriever("schema:actor").contents,
    registry=MINCORE_REGISTRY,
    format_checker=Draft202012Validator.FORMAT_CHECKER,
)


async def get_remote_actor(actor_uri: str):
    # NOTE No authn/authz
    async with httpx.AsyncClient() as client:
        response = await client.get(actor_uri)
        response.raise_for_status()
        # TODO Validate actor with JSON Schema
        actor = response.json()
        ACTOR_VALIDATOR.validate(actor)
        return actor


async def get_remote_actor_inbox(actor_uri: str):
    return (await get_remote_actor(actor_uri))["inbox"]

activitypub_mincore/actor.py
steve-bate/activitypub-mincore
import asyncio
import logging
import uuid

import fastapi
import httpx

from activitypub_mincore.actor import get_local_actor, get_remote_actor_inbox
from activitypub_mincore.support.validation import (
    MINCORE_ACTIVITY_TYPES,
    create_activity_validator,
    validate_activity,
)

logger = logging.getLogger("follower")

app = fastapi.FastAPI()

EXT_ACTIVITY_VALIDATOR = create_activity_validator(
    types=MINCORE_ACTIVITY_TYPES + ["Create"]
)


@app.post("/{path:path}")
async def post__inbox(request: fastapi.Request):
    instance_actor = get_local_actor()
    if str(request.url) != instance_actor["inbox"]:
        raise fastapi.HTTPException(403, "Forbidden")
    try:
        activity = await request.json()
        logger.info(f"Received activity: {activity}")
    except httpx.HTTPError as httpEx:
        raise httpEx
    except Exception as ex:  # noqa
        logger.exception(ex)
        raise fastapi.HTTPException(500, "Internal Server Error")


async def follow(uri_to_follow: str, *, ntries: int = 1):
    while ntries:
        logger.info(f"Requesting to Follow {uri_to_follow}")
        try:
            remote_inbox = await get_remote_actor_inbox(uri_to_follow)
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    remote_inbox,
                    # "Pre-flight" validation
                    json=validate_activity(
                        {
                            # Must have an id for the response
                            "id": f"{get_local_actor()['id']}/{uuid.uuid4()}",
                            "type": "Follow",
                            "actor": get_local_actor()["id"],
                            "object": uri_to_follow,
                        },
                        EXT_ACTIVITY_VALIDATOR,
                    ),
                )
                response.raise_for_status()
        except httpx.ConnectError:
            logging.warning("Connection failed, retrying...")
            ntries -= 1
            if ntries > 0:
                await asyncio.sleep(2)
                continue
        except Exception as ex:
            logger.exception(ex)
        break


async def send_follow(uri_to_follow: str, delay: int = 2):
    logger.info(f"Waiting to send follow for {uri_to_follow}")
    await asyncio.sleep(delay)
    logger.info(f"Sent Follow request for {uri_to_follow}")
    await follow(uri_to_follow, ntries=10000)

activitypub_mincore/follower.py
steve-bate/activitypub-mincore
import asyncio
import logging
from datetime import datetime
from typing import Any

import fastapi
import httpx
from bidict import bidict

from activitypub_mincore.actor import get_local_actor, get_remote_actor_inbox
from activitypub_mincore.support.validation import (
    MINCORE_ACTIVITY_TYPES,
    create_activity_validator,
    validate_activity,
)

logger = logging.getLogger("publisher")

app = fastapi.FastAPI()

_followers = bidict()

EXT_ACTIVITY_VALIDATOR = create_activity_validator(
    types=MINCORE_ACTIVITY_TYPES + ["Create"]
)


async def handle_follow(local_actor: dict[str, Any], activity: dict[str, Any]) -> None:
    async def send_follow_response(type_: str):
        response = await client.post(
            follower_inbox,
            json=validate_activity(
                {
                    # transient, so no id
                    "type": type_,
                    "actor": local_actor["id"],
                    "object": activity.get("id"),
                },
                EXT_ACTIVITY_VALIDATOR,
            ),
        )
        response.raise_for_status()

    follower_uri = activity.get("actor")
    if isinstance(follower_uri, str):
        async with httpx.AsyncClient() as client:
            follower_inbox = await get_remote_actor_inbox(follower_uri)
            if follower_inbox in _followers:
                await send_follow_response("Reject")
                logger.info(f"Rejected: already following {follower_uri}")
            else:
                await send_follow_response("Accept")
                logger.info(f"Accepted Follow from {follower_uri}")
            _followers[follower_inbox] = activity.get("id")
    else:
        logger.warn(f"Follower URI must be a string: {follower_uri}")


def handle_undo_follow(activity: dict[str, Any]) -> None:
    follow_activity_uri = activity.get("object")
    if isinstance(follow_activity_uri, str):
        follower_inbox = _followers.inverse.get(follow_activity_uri)
        if follower_inbox:
            del _followers[follower_inbox]
            logger.info(f"Follower inbox removed: {follower_inbox}")
        else:
            logger.warn(f"Uknown follow activity: {follow_activity_uri}")
    else:
        logger.warn(f"Follow activity must be a string (URI): {activity}")


@app.post("/{path:path}")
async def post__inbox(request: fastapi.Request):
    instance_actor = get_local_actor()
    if str(request.url) != instance_actor["inbox"]:
        raise fastapi.HTTPException(403, "Forbidden")
    try:
        activity = await request.json()
        match activity.get("type"):
            case "Follow":
                await handle_follow(instance_actor, activity)
            case "Undo":
                handle_undo_follow(activity)
            case _:
                raise fastapi.HTTPException(400, "Bad request")
    except httpx.HTTPError as httpEx:
        raise httpEx
    except Exception as ex:  # noqa
        logger.exception(ex)
        raise fastapi.HTTPException(500, "Internal Server Error")


async def publish_once():
    if _followers:
        inboxes = list(_followers)  # copy, to allow mutation
        logger.info(f"publishing to {inboxes}")
        activity = validate_activity(
            {
                # transient objects, no ids
                "type": "Create",
                "actor": get_local_actor()["id"],
                "object": {
                    "type": "Note",
                    "content": f"The time is {datetime.now().isoformat()}",
                    "to": inboxes,
                },
            },
            EXT_ACTIVITY_VALIDATOR,
        )
        for inbox_uri in inboxes:
            try:
                async with httpx.AsyncClient() as client:
                    response = await client.post(inbox_uri, json=activity)
                    response.raise_for_status()
            except Exception as ex:
                # Remove follower if there is an error
                if not isinstance(ex, httpx.ConnectError):
                    logger.exception(ex)
                else:
                    logger.error(ex)
                logger.warning(f"Removing {inbox_uri} from followers")
                _followers.pop(inbox_uri)


async def publish():
    try:
        while True:
            await publish_once()
            await asyncio.sleep(5)
    except OSError as ex:
        logging.warning(ex)
    except Exception as ex:
        logging.exception(ex)
    finally:
        logger.warning("publish task exiting")

activitypub_mincore/publisher.py
steve-bate/activitypub-mincore
import asyncio

import click
import uvicorn

from activitypub_mincore import follower, publisher
from activitypub_mincore.actor import initialize_actor
from activitypub_mincore.support.server import Server


@click.group
def cli():
    """ActivityPub Minimal Core"""


@cli.command("publisher")
@click.option("--port", help="server port", default=8000)
def publisher_instance(port: int):
    """Publish-only, single-actor instance"""
    config = uvicorn.Config(app=publisher.app, port=port, log_config=None)
    server = Server(config, [publisher.publish()])
    initialize_actor(server)
    asyncio.run(server.run())


@cli.command("follower")
@click.argument("tofollow", nargs=-1, metavar="ACTOR_URIS...")
@click.option("--port", metavar="PORT", type=int, help="server port", default=8001)
def follower_instance(tofollow: list[str], port: int):
    """Follow-only, single actor instance"""
    if not tofollow:
        tofollow = ["http://127.0.0.1:8000/actor"]
    config = uvicorn.Config(app=follower.app, port=port, log_config=None)
    server = Server(config, [follower.send_follow(uri) for uri in tofollow])
    initialize_actor(server)
    asyncio.run(server.run())


#

activitypub_mincore/support/cli.py
steve-bate/activitypub-mincore
import asyncio
import logging
import sys
from typing import Awaitable

import uvicorn

logging.basicConfig(
    stream=sys.stdout,
    level=logging.INFO,
    format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
)


class Server(uvicorn.Server):
    def __init__(
        self, config: uvicorn.Config, background_tasks: list[Awaitable] | None = None
    ):
        logging.getLogger("uvicorn.error").name = "uvicorn"
        super().__init__(config)
        self.background_tasks = background_tasks or []
        self.tasks: list[asyncio.Task] = []

    def handle_exit(self, sig: int, frame) -> None:
        if self.tasks:
            for task in self.tasks:
                if task != self.server_task:
                    task.cancel()
        return super().handle_exit(sig, frame)

    async def run(self):
        self.server_task = asyncio.create_task(self.serve())
        self.tasks = [self.server_task] + [
            asyncio.create_task(t) for t in self.background_tasks
        ]
        await asyncio.wait(self.tasks)

activitypub_mincore/support/server.py
steve-bate/activitypub-mincore
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

import referencing
import referencing.retrieval
from jsonschema import Draft202012Validator
from jsonschema.protocols import Validator
from referencing.jsonschema import DRAFT202012

SCHEMA_DIR = Path(__file__).parent.parent / "schemas"


@referencing.retrieval.to_cached_resource()
def schema_retriever(uri):
    url = urlparse(uri)
    filepath = SCHEMA_DIR / f"{url.path}-schema.json"
    with open(filepath) as fp:
        return fp.read()


MINCORE_REGISTRY = referencing.Registry(retrieve=schema_retriever)

MINCORE_ACTIVITY_TYPES = ["Follow", "Accept", "Reject"]


def create_activity_validator(
    *, root_schema: str = "schema:activity", types=None
) -> Validator:
    types = set(types or MINCORE_ACTIVITY_TYPES)
    return Draft202012Validator(
        schema_retriever(root_schema).contents,
        registry=MINCORE_REGISTRY.with_resource(
            uri="schema:known-activity-types",
            resource=referencing.Resource(
                contents={
                    "$schema": "https://json-schema.org/draft/2020-12/schema",
                    "$id": "schema:known-activities",
                    "oneOf": [{"const": a} for a in types],
                },
                specification=DRAFT202012,
            ),
        ),
        format_checker=Draft202012Validator.FORMAT_CHECKER,
    )


MINCORE_ACTIVITY_VALIDATOR = create_activity_validator()


def validate_activity(activity: dict[str, Any], validator: Validator) -> dict[str, Any]:
    """Validates an activity and returns it, if valid."""
    validator.validate(activity)
    return activity

activitypub_mincore/support/validation.py
steve-bate/activitypub-mincore
import pytest
from fastapi.testclient import TestClient

from activitypub_mincore import actor
from tests.support import MockServer


@pytest.fixture
def mock_server(test_client, monkeypatch, mock_actor):
    server = MockServer(test_client, monkeypatch)
    server.add_response(mock_actor["id"], mock_actor)
    return server


@pytest.fixture
def mock_actor():
    mock_actor_uri = "http://server.test/actor"
    profile = {
        "id": mock_actor_uri,
        "type": "Service",
        "inbox": f"{mock_actor_uri}/inbox",
        "outbox": f"{mock_actor_uri}/outbox",
    }
    return profile


@pytest.fixture
def test_client(app):
    app.include_router(actor.router)
    with TestClient(app) as client:
        # actor under test
        actor_uri = f"{client.base_url}/actor"
        actor._actor = {
            "id": actor_uri,
            "type": "Service",
            "inbox": f"{actor_uri}/inbox",
            "outbox": f"{actor_uri}/outbox",
        }
        yield client

tests/conftest.py
steve-bate/activitypub-mincore
import json
from dataclasses import dataclass, field
from typing import Any

import httpx
from fastapi.testclient import TestClient


@dataclass
class MockRequest:
    method: str
    url: str
    options: dict


@dataclass
class MockResponse:
    content: bytes
    method: str = field(default="get")
    status_code: int = field(default=200)


class MockServer:
    def __init__(self, test_client: TestClient, monkeypatch):
        self.requests: list[MockRequest] = []
        self._get_responses: dict[str, MockResponse] = {}
        self._post_responses: dict[str, MockResponse] = {}
        self._test_client = test_client
        self._test_client_complete = False
        monkeypatch.setattr(httpx.AsyncClient, "__aenter__", self)

    def _ensure_client_completion(self):
        if not self._test_client_complete:
            self._test_client.__exit__()
            self._test_client_complete = True

    def add_response(
        self,
        url: str,
        data: str | bytes | dict,
        method: str = "get",
        status_code: int = 200,
    ):
        if isinstance(data, str):
            data = data.encode()
        elif isinstance(data, dict):
            data = json.dumps(data).encode()
        response = MockResponse(
            method=method.lower(),
            content=data,
            status_code=status_code,
        )
        responses = self._get_responses if method == "get" else self._post_responses
        responses[url] = response

    @staticmethod
    def _httpx_error_response(
        method: str, url: str, status_code: int
    ) -> httpx.Response:
        httpx_response = httpx.Response(
            404, stream=httpx._transports.default.ResponseStream([])
        )
        httpx_response.request = httpx.Request(method, url)
        return httpx_response

    def sync_get(self, args, kwargs):
        self.requests.append(MockRequest(method="get", url=args[0], options=kwargs))
        mock_response = self._get_responses.get(args[0])
        if mock_response:
            # # May support a "count" in the future
            # del self._get_responses[args[0]]
            return self._to_httpx_response(args[0], mock_response)
        else:
            return self._httpx_error_response("get", args[0], 404)

    async def get(self, *args, **kwargs):
        return self.sync_get(args, kwargs)

    @staticmethod
    def _to_httpx_response(url: str, mock_response: MockResponse):
        payload = mock_response.content
        httpx_response = httpx.Response(
            mock_response.status_code,
            stream=httpx._transports.default.ResponseStream([payload]),
        )
        httpx_response.request = httpx.Request(mock_response.method, url)
        setattr(httpx_response, "_content", payload)
        return httpx_response

    def sync_post(self, args, kwargs):
        url = args[0]
        self.requests.append(
            MockRequest(
                method="post",
                url=url,
                options=kwargs,
            )
        )
        mock_response = self._post_responses.get(url)
        if mock_response:
            return self._to_response(url, mock_response)
        else:
            response = httpx.Response(status_code=200)
            response.request = httpx.Request("post", url)
            return response

    async def post(self, *args, **kwargs):
        return self.sync_post(args, kwargs)

    def received_post(self, pattern: dict[str, Any]) -> MockRequest | None:
        for request in self.requests:
            if request.method == "post":
                if "json" in request.options:
                    payload = request.options["json"]
                    matched = True
                    for k, v in pattern.items():
                        if k not in payload or payload[k] != v:
                            matched = False
                            break
                    if matched:
                        return request
        return None

    # __aenter callback
    async def __call__(self, *args):
        return self

tests/support.py
steve-bate/activitypub-mincore
import pytest
from fastapi.testclient import TestClient

from activitypub_mincore.follower import app as follower_app
from activitypub_mincore.follower import follow
from tests.support import MockServer


@pytest.fixture
def app():
    return follower_app


async def test_following(
    test_client: TestClient, mock_actor: dict, mock_server: MockServer
):
    # trigger follow request
    await follow(mock_actor["id"])
    assert mock_server.received_post({"type": "Follow", "object": mock_actor["id"]})
    # simulate publication (smoke test)
    response = test_client.get(f"{test_client.base_url}/actor")
    assert response.is_success
    actor = response.json()
    inbox = actor["inbox"]
    response = test_client.post(
        inbox,
        json={
            "id": "http://server.test/create",
            "type": "Create",
            "actor": mock_actor["id"],
            "to": actor["id"],
            "object": {"type": "Note", "content": "Hello"},
        },
    )
    assert response.is_success

tests/test_follower.py
steve-bate/activitypub-mincore
import pytest
from fastapi.testclient import TestClient

from activitypub_mincore.publisher import app as publisher_app
from activitypub_mincore.publisher import publish_once
from tests.support import MockServer


@pytest.fixture
def app():
    return publisher_app


async def test_follow_handling(
    test_client: TestClient, mock_actor: dict, mock_server: MockServer
):
    response = test_client.get(f"{test_client.base_url}/actor")
    assert response.is_success
    # post Follow request
    actor = response.json()
    inbox = actor["inbox"]
    follow_activity_uri = "http://server.test/follow"
    response = test_client.post(
        inbox,
        json={
            "id": follow_activity_uri,
            "type": "Follow",
            "actor": mock_actor["id"],
            "object": actor["id"],
        },
    )
    assert response.is_success
    assert mock_server.received_post({"type": "Accept"}), "No accept"
    # publish
    await publish_once()
    assert mock_server.received_post({"type": "Create"}), "No Create"
    # # handle unfollow (Undo/Follow)
    # mock_server.requests.clear()
    # response = test_client.post(
    #     inbox,
    #     json={
    #         "id": "http://server.test/unfollow",
    #         "type": "Undo",
    #         "actor": mock_actor["id"],
    #         "object": follow_activity_uri,
    #     },
    # )
    # assert response.is_success
    # # publish again, should not receive this one
    # await publish_once()
    # assert mock_server.received_post({"type": "Create"}) is None, "Unexpected Create"

tests/test_publisher.py
steve-bate/activitypub-mincore
from pathlib import Path

import pytest
from jsonschema import ValidationError

from activitypub_mincore.support.validation import MINCORE_ACTIVITY_VALIDATOR

SCHEMA_DIR = Path(__file__).parent.parent / "activitypub_mincore" / "schemas"


VALIDATION_TEST_CASES = [
    # Wrong type type
    ({"type": 123}, "fail"),
    # Wrong type type
    ({"type": ["Follow"]}, "fail"),
    # Invalid type
    ({"type": "Bogus"}, "fail"),
    # No object
    ({"type": "Follow"}, "fail"),
    # Accepts objects for "object"
    ({"type": "Follow", "object": {"id": "http://server.test"}}, "pass"),
    # Invalid URI
    ({"type": "Follow", "object": "@foo@bar.test"}, "fail"),
    # Valid URI
    ({"type": "Follow", "object": "https://server.test/"}, "pass"),
    ({"type": "Accept", "object": "https://server.test/"}, "pass"),
    ({"type": "Reject", "object": "https://server.test/"}, "pass"),
    # Extra properties
    ({"type": "Follow", "object": "https://server.test/", "EXTRA": "extra"}, "pass"),
    # Mispelled "actor"... passes as extra property
    (
        {"type": "Follow", "object": "https://server.test/", "acter": "Mel Brooks"},
        "pass",
    ),
]


@pytest.mark.parametrize(["activity", "outcome"], VALIDATION_TEST_CASES)
def test_activity_validation(activity: dict, outcome: str):
    try:
        MINCORE_ACTIVITY_VALIDATOR.validate(activity)
        assert outcome == "pass", lambda: f"Unexpected pass: {activity}"
    except ValidationError as ex:
        assert outcome == "fail", ex.message

tests/test_validation.py
steve-bate/activitypub-mincore