rose-ash/shared/infrastructure/ap_inbox_handlers.py

"""Reusable AP inbox handlers for all apps.

Extracted from federation/bp/actors/routes.py so that every app's
shared AP blueprint can process Follow, Undo, Accept, Create, etc.
"""
from __future__ import annotations

import json
import logging
import uuid
from datetime import datetime, timezone

import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from shared.models.federation import (
    ActorProfile, APInboxItem, APInteraction, APNotification,
    APRemotePost, APActivity, RemoteActor,
)
from shared.services.registry import services

log = logging.getLogger(__name__)

AP_CONTENT_TYPE = "application/activity+json"


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _is_safe_url(url: str) -> bool:
    """Reject URLs pointing to private/internal IPs to prevent SSRF."""
    from urllib.parse import urlparse
    import ipaddress

    parsed = urlparse(url)

    # Require HTTPS
    if parsed.scheme != "https":
        return False

    hostname = parsed.hostname
    if not hostname:
        return False

    # Block obvious internal hostnames
    if hostname in ("localhost", "127.0.0.1", "::1", "0.0.0.0"):
        return False

    try:
        addr = ipaddress.ip_address(hostname)
        if addr.is_private or addr.is_loopback or addr.is_reserved or addr.is_link_local:
            return False
    except ValueError:
        # Not an IP literal — hostname is fine (DNS resolution handled by httpx)
        # Block common internal DNS patterns
        if hostname.endswith(".internal") or hostname.endswith(".local"):
            return False

    return True


async def fetch_remote_actor(actor_url: str) -> dict | None:
    """Fetch a remote actor's JSON-LD profile."""
    if not _is_safe_url(actor_url):
        log.warning("Blocked SSRF attempt: %s", actor_url)
        return None
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            resp = await client.get(
                actor_url,
                headers={"Accept": AP_CONTENT_TYPE},
            )
            if resp.status_code == 200:
                return resp.json()
    except Exception:
        log.exception("Failed to fetch remote actor: %s", actor_url)
    return None


async def send_accept(
    actor: ActorProfile,
    follow_activity: dict,
    follower_inbox: str,
    domain: str,
) -> None:
    """Send an Accept activity back to the follower."""
    from shared.utils.http_signatures import sign_request
    from urllib.parse import urlparse

    username = actor.preferred_username
    actor_url = f"https://{domain}/users/{username}"

    accept_id = f"{actor_url}/activities/{uuid.uuid4()}"
    accept = {
        "@context": "https://www.w3.org/ns/activitystreams",
        "id": accept_id,
        "type": "Accept",
        "actor": actor_url,
        "object": follow_activity,
    }

    body_bytes = json.dumps(accept).encode()
    key_id = f"{actor_url}#main-key"

    parsed = urlparse(follower_inbox)
    headers = sign_request(
        private_key_pem=actor.private_key_pem,
        key_id=key_id,
        method="POST",
        path=parsed.path,
        host=parsed.netloc,
        body=body_bytes,
    )
    headers["Content-Type"] = AP_CONTENT_TYPE

    log.info("Accept payload → %s: %s", follower_inbox, json.dumps(accept)[:500])

    try:
        async with httpx.AsyncClient(timeout=15) as client:
            resp = await client.post(
                follower_inbox,
                content=body_bytes,
                headers=headers,
            )
            log.info("Accept → %s: %d %s", follower_inbox, resp.status_code, resp.text[:200])
    except Exception:
        log.exception("Failed to send Accept to %s", follower_inbox)


async def backfill_follower(
    session: AsyncSession,
    actor: ActorProfile,
    follower_inbox: str,
    domain: str,
    origin_app: str | None = None,
) -> None:
    """Deliver recent *current* Create activities to a new follower's inbox.

    Skips Creates whose source was later Deleted, and uses the latest
    Update data when available (so the follower sees the current version).
    """
    from shared.events.handlers.ap_delivery_handler import (
        _build_activity_json, _deliver_to_inbox,
    )

    filters = [
        APActivity.actor_profile_id == actor.id,
        APActivity.is_local == True,  # noqa: E712
        APActivity.activity_type == "Create",
        APActivity.source_type.isnot(None),
        APActivity.source_id.isnot(None),
    ]
    if origin_app is not None:
        filters.append(APActivity.origin_app == origin_app)

    creates = (
        await session.execute(
            select(APActivity).where(*filters)
            .order_by(APActivity.published.desc())
            .limit(40)
        )
    ).scalars().all()

    if not creates:
        return

    # Collect source keys that have been Deleted
    source_keys = {(c.source_type, c.source_id) for c in creates}
    deleted_keys: set[tuple[str | None, int | None]] = set()
    if source_keys:
        deletes = (
            await session.execute(
                select(APActivity.source_type, APActivity.source_id).where(
                    APActivity.actor_profile_id == actor.id,
                    APActivity.activity_type == "Delete",
                    APActivity.is_local == True,  # noqa: E712
                )
            )
        ).all()
        deleted_keys = {(d[0], d[1]) for d in deletes}

    # For sources with Updates, grab the latest Update's object_data
    updated_data: dict[tuple[str | None, int | None], dict] = {}
    if source_keys:
        updates = (
            await session.execute(
                select(APActivity).where(
                    APActivity.actor_profile_id == actor.id,
                    APActivity.activity_type == "Update",
                    APActivity.is_local == True,  # noqa: E712
                ).order_by(APActivity.published.desc())
            )
        ).scalars().all()
        for u in updates:
            key = (u.source_type, u.source_id)
            if key not in updated_data and key in source_keys:
                updated_data[key] = u.object_data or {}

    # Filter to current, non-deleted Creates (limit 20)
    activities = []
    for c in creates:
        key = (c.source_type, c.source_id)
        if key in deleted_keys:
            continue
        # Apply latest Update data if available
        if key in updated_data:
            c.object_data = updated_data[key]
        activities.append(c)
        if len(activities) >= 20:
            break

    if not activities:
        return

    log.info(
        "Backfilling %d posts to %s for @%s",
        len(activities), follower_inbox, actor.preferred_username,
    )

    async with httpx.AsyncClient() as client:
        for activity in reversed(activities):  # oldest first
            activity_json = _build_activity_json(activity, actor, domain)
            await _deliver_to_inbox(client, follower_inbox, activity_json, actor, domain)


# ---------------------------------------------------------------------------
# Inbox activity handlers
# ---------------------------------------------------------------------------

async def handle_follow(
    session: AsyncSession,
    actor_row: ActorProfile,
    body: dict,
    from_actor_url: str,
    domain: str,
    app_domain: str = "federation",
) -> None:
    """Process a Follow activity: add follower, send Accept, backfill."""
    remote_actor = await fetch_remote_actor(from_actor_url)
    if not remote_actor:
        log.warning("Could not fetch remote actor for Follow: %s", from_actor_url)
        return

    follower_inbox = remote_actor.get("inbox")
    if not follower_inbox:
        log.warning("Remote actor has no inbox: %s", from_actor_url)
        return

    remote_username = remote_actor.get("preferredUsername", "")
    from urllib.parse import urlparse
    remote_domain = urlparse(from_actor_url).netloc
    follower_acct = f"{remote_username}@{remote_domain}" if remote_username else from_actor_url

    pub_key = (remote_actor.get("publicKey") or {}).get("publicKeyPem")

    await services.federation.add_follower(
        session,
        actor_row.preferred_username,
        follower_acct=follower_acct,
        follower_inbox=follower_inbox,
        follower_actor_url=from_actor_url,
        follower_public_key=pub_key,
        app_domain=app_domain,
    )

    log.info(
        "New follower: %s → @%s (app_domain=%s)",
        follower_acct, actor_row.preferred_username, app_domain,
    )

    # Notification
    ra = (
        await session.execute(
            select(RemoteActor).where(RemoteActor.actor_url == from_actor_url)
        )
    ).scalar_one_or_none()
    if not ra:
        ra_dto = await services.federation.get_or_fetch_remote_actor(session, from_actor_url)
        if ra_dto:
            ra = (await session.execute(
                select(RemoteActor).where(RemoteActor.actor_url == from_actor_url)
            )).scalar_one_or_none()

    if ra:
        notif = APNotification(
            actor_profile_id=actor_row.id,
            notification_type="follow",
            from_remote_actor_id=ra.id,
            app_domain=app_domain,
        )
        session.add(notif)

    # Send Accept
    await send_accept(actor_row, body, follower_inbox, domain)

    # Backfill: deliver recent posts (filtered by origin_app for per-app follows)
    backfill_origin = app_domain if app_domain != "federation" else None
    await backfill_follower(session, actor_row, follower_inbox, domain, origin_app=backfill_origin)


async def handle_undo(
    session: AsyncSession,
    actor_row: ActorProfile,
    body: dict,
    from_actor_url: str,
    app_domain: str = "federation",
) -> None:
    """Process an Undo activity (typically Undo Follow)."""
    inner = body.get("object")
    if not inner:
        return

    inner_type = inner.get("type") if isinstance(inner, dict) else None
    if inner_type == "Follow":
        from urllib.parse import urlparse
        remote_domain = urlparse(from_actor_url).netloc
        remote_actor = await fetch_remote_actor(from_actor_url)
        remote_username = ""
        if remote_actor:
            remote_username = remote_actor.get("preferredUsername", "")
        follower_acct = f"{remote_username}@{remote_domain}" if remote_username else from_actor_url

        removed = await services.federation.remove_follower(
            session, actor_row.preferred_username, follower_acct,
            app_domain=app_domain,
        )
        if removed:
            log.info("Unfollowed: %s → @%s (app_domain=%s)", follower_acct, actor_row.preferred_username, app_domain)
        else:
            log.debug("Undo Follow: follower not found: %s", follower_acct)
    else:
        log.debug("Undo for %s — not handled", inner_type)


async def handle_accept(
    session: AsyncSession,
    actor_row: ActorProfile,
    body: dict,
    from_actor_url: str,
) -> None:
    """Process Accept activity — update outbound follow state."""
    inner = body.get("object")
    if not inner:
        return

    inner_type = inner.get("type") if isinstance(inner, dict) else None
    if inner_type == "Follow":
        await services.federation.accept_follow_response(
            session, actor_row.preferred_username, from_actor_url,
        )
        log.info("Follow accepted by %s for @%s", from_actor_url, actor_row.preferred_username)


async def handle_create(
    session: AsyncSession,
    actor_row: ActorProfile,
    body: dict,
    from_actor_url: str,
    federation_domain: str,
) -> None:
    """Process Create(Note/Article) — ingest remote post."""
    obj = body.get("object")
    if not obj or not isinstance(obj, dict):
        return

    obj_type = obj.get("type", "")
    if obj_type not in ("Note", "Article"):
        log.debug("Create with type %s — skipping", obj_type)
        return

    remote = await services.federation.get_or_fetch_remote_actor(session, from_actor_url)
    if not remote:
        log.warning("Could not resolve remote actor for Create: %s", from_actor_url)
        return

    await services.federation.ingest_remote_post(session, remote.id, body, obj)
    log.info("Ingested %s from %s", obj_type, from_actor_url)

    # Mention notification
    tags = obj.get("tag", [])
    if isinstance(tags, list):
        for tag in tags:
            if not isinstance(tag, dict):
                continue
            if tag.get("type") != "Mention":
                continue
            href = tag.get("href", "")
            if f"https://{federation_domain}/users/" in href:
                mentioned_username = href.rsplit("/", 1)[-1]
                mentioned = await services.federation.get_actor_by_username(
                    session, mentioned_username,
                )
                if mentioned:
                    rp = (await session.execute(
                        select(APRemotePost).where(
                            APRemotePost.object_id == obj.get("id")
                        )
                    )).scalar_one_or_none()

                    ra = (await session.execute(
                        select(RemoteActor).where(RemoteActor.actor_url == from_actor_url)
                    )).scalar_one_or_none()

                    notif = APNotification(
                        actor_profile_id=mentioned.id,
                        notification_type="mention",
                        from_remote_actor_id=ra.id if ra else None,
                        target_remote_post_id=rp.id if rp else None,
                    )
                    session.add(notif)

    # Reply notification
    in_reply_to = obj.get("inReplyTo")
    if in_reply_to and f"https://{federation_domain}/users/" in str(in_reply_to):
        local_activity = (await session.execute(
            select(APActivity).where(
                APActivity.activity_id == in_reply_to,
            )
        )).scalar_one_or_none()
        if local_activity:
            ra = (await session.execute(
                select(RemoteActor).where(RemoteActor.actor_url == from_actor_url)
            )).scalar_one_or_none()
            rp = (await session.execute(
                select(APRemotePost).where(
                    APRemotePost.object_id == obj.get("id")
                )
            )).scalar_one_or_none()

            notif = APNotification(
                actor_profile_id=local_activity.actor_profile_id,
                notification_type="reply",
                from_remote_actor_id=ra.id if ra else None,
                target_remote_post_id=rp.id if rp else None,
            )
            session.add(notif)


async def handle_update(
    session: AsyncSession,
    actor_row: ActorProfile,
    body: dict,
    from_actor_url: str,
) -> None:
    """Process Update — re-ingest remote post."""
    obj = body.get("object")
    if not obj or not isinstance(obj, dict):
        return
    obj_type = obj.get("type", "")
    if obj_type in ("Note", "Article"):
        remote = await services.federation.get_or_fetch_remote_actor(session, from_actor_url)
        if remote:
            await services.federation.ingest_remote_post(session, remote.id, body, obj)
            log.info("Updated %s from %s", obj_type, from_actor_url)


async def handle_delete(
    session: AsyncSession,
    actor_row: ActorProfile,
    body: dict,
    from_actor_url: str,
) -> None:
    """Process Delete — remove remote post."""
    obj = body.get("object")
    if isinstance(obj, str):
        object_id = obj
    elif isinstance(obj, dict):
        object_id = obj.get("id", "")
    else:
        return
    if object_id:
        await services.federation.delete_remote_post(session, object_id)
        log.info("Deleted remote post %s from %s", object_id, from_actor_url)


async def handle_like(
    session: AsyncSession,
    actor_row: ActorProfile,
    body: dict,
    from_actor_url: str,
) -> None:
    """Process incoming Like — record interaction + notify."""
    object_id = body.get("object", "")
    if isinstance(object_id, dict):
        object_id = object_id.get("id", "")
    if not object_id:
        return

    remote = await services.federation.get_or_fetch_remote_actor(session, from_actor_url)
    if not remote:
        return

    ra = (await session.execute(
        select(RemoteActor).where(RemoteActor.actor_url == from_actor_url)
    )).scalar_one_or_none()

    target = (await session.execute(
        select(APActivity).where(APActivity.activity_id == object_id)
    )).scalar_one_or_none()

    if not target:
        log.info("Like from %s for %s (target not found locally)", from_actor_url, object_id)
        return

    interaction = APInteraction(
        remote_actor_id=ra.id if ra else None,
        post_type="local",
        post_id=target.id,
        interaction_type="like",
        activity_id=body.get("id"),
    )
    session.add(interaction)

    notif = APNotification(
        actor_profile_id=target.actor_profile_id,
        notification_type="like",
        from_remote_actor_id=ra.id if ra else None,
        target_activity_id=target.id,
    )
    session.add(notif)
    log.info("Like from %s on activity %s", from_actor_url, object_id)


async def handle_announce(
    session: AsyncSession,
    actor_row: ActorProfile,
    body: dict,
    from_actor_url: str,
) -> None:
    """Process incoming Announce (boost) — record interaction + notify."""
    object_id = body.get("object", "")
    if isinstance(object_id, dict):
        object_id = object_id.get("id", "")
    if not object_id:
        return

    remote = await services.federation.get_or_fetch_remote_actor(session, from_actor_url)
    if not remote:
        return

    ra = (await session.execute(
        select(RemoteActor).where(RemoteActor.actor_url == from_actor_url)
    )).scalar_one_or_none()

    target = (await session.execute(
        select(APActivity).where(APActivity.activity_id == object_id)
    )).scalar_one_or_none()

    if not target:
        log.info("Announce from %s for %s (target not found locally)", from_actor_url, object_id)
        return

    interaction = APInteraction(
        remote_actor_id=ra.id if ra else None,
        post_type="local",
        post_id=target.id,
        interaction_type="boost",
        activity_id=body.get("id"),
    )
    session.add(interaction)

    notif = APNotification(
        actor_profile_id=target.actor_profile_id,
        notification_type="boost",
        from_remote_actor_id=ra.id if ra else None,
        target_activity_id=target.id,
    )
    session.add(notif)
    log.info("Announce from %s on activity %s", from_actor_url, object_id)


async def dispatch_inbox_activity(
    session: AsyncSession,
    actor_row: ActorProfile,
    body: dict,
    from_actor_url: str,
    domain: str,
    app_domain: str = "federation",
) -> None:
    """Route an inbox activity to the correct handler."""
    activity_type = body.get("type", "")

    if activity_type == "Follow":
        await handle_follow(session, actor_row, body, from_actor_url, domain, app_domain=app_domain)
    elif activity_type == "Undo":
        await handle_undo(session, actor_row, body, from_actor_url, app_domain=app_domain)
    elif activity_type == "Accept":
        await handle_accept(session, actor_row, body, from_actor_url)
    elif activity_type == "Create":
        await handle_create(session, actor_row, body, from_actor_url, domain)
    elif activity_type == "Update":
        await handle_update(session, actor_row, body, from_actor_url)
    elif activity_type == "Delete":
        await handle_delete(session, actor_row, body, from_actor_url)
    elif activity_type == "Like":
        await handle_like(session, actor_row, body, from_actor_url)
    elif activity_type == "Announce":
        await handle_announce(session, actor_row, body, from_actor_url)