#!/usr/bin/env python3

"""Which applications may use the camera and microphone.

Read from xdg-desktop-portal's permission store, which is where an application
that asks through the portal has its answer recorded. That is the whole of what
this can control, and the limit is worth stating plainly rather than implying a
protection that does not exist: a native binary opens /dev/video0 directly and
no desktop setting stands in its way. What this covers is Flatpaks and anything
else that goes through the portal -- which on this machine is most of what would
ever ask.

Devices with no recorded application are reported as empty rather than omitted,
so the page can say "nothing has asked" instead of showing nothing at all.

    panama-permissions snapshot
    panama-permissions set DEVICE APP_ID allow|deny
    panama-permissions forget DEVICE APP_ID
"""

from __future__ import annotations

import json
import re
import subprocess
import sys

TABLE = "devices"

# The devices the portal arbitrates. Listed rather than discovered so a device
# nothing has asked for still appears, which is the difference between "no
# application uses your microphone" and a page that silently omits it.
DEVICES = (
    ("camera", "Camera"),
    ("microphone", "Microphone"),
    ("speakers", "Speakers"),
)

APP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
DEVICE_ID = re.compile(r"^[a-z]+$")

ALLOWED = "yes"
DENIED = "no"


class BoundaryError(RuntimeError):
    """A user-visible validation or permission-store failure."""


def run(command: list[str], timeout: float = 15.0) -> subprocess.CompletedProcess:
    try:
        return subprocess.run(command, capture_output=True, text=True, timeout=timeout)
    except FileNotFoundError as error:
        raise BoundaryError("busctl is not available.") from error
    except subprocess.TimeoutExpired as error:
        raise BoundaryError("The permission store did not respond.") from error


def portal_call(method: str, signature: str, *arguments: str) -> dict | None:
    """One call to the permission store, as JSON.

    --json=short rather than busctl's text output, which escapes non-ASCII into
    octal and would mangle an application name.
    """
    result = run([
        "busctl", "--user", "--json=short", "call",
        "org.freedesktop.impl.portal.PermissionStore",
        "/org/freedesktop/impl/portal/PermissionStore",
        "org.freedesktop.impl.portal.PermissionStore",
        method, signature, *arguments,
    ])
    if result.returncode != 0:
        detail = (result.stderr or "").strip()
        # A device nothing has ever asked for has no row at all, and the store
        # says so as "No entry for camera". Matched on the store's actual words
        # rather than a guess at them.
        lowered = detail.lower()
        if "no entry" in lowered or "not found" in lowered:
            return None
        raise BoundaryError(detail.splitlines()[-1] if detail else "The permission store refused that.")
    try:
        return json.loads(result.stdout or "null")
    except json.JSONDecodeError as error:
        raise BoundaryError("The permission store returned something unreadable.") from error


def available() -> bool:
    result = run(["busctl", "--user", "list"])
    return "org.freedesktop.impl.portal.PermissionStore" in (result.stdout or "")


def entries_for(device: str) -> list[dict]:
    payload = portal_call("Lookup", "ss", TABLE, device)
    if not payload:
        return []
    data = payload.get("data") or []
    if not data or not isinstance(data[0], dict):
        return []

    found = []
    for app_id, permissions in data[0].items():
        values = [str(value) for value in (permissions or [])]
        found.append({
            "app": app_id,
            # Anything that is not an explicit "yes" is treated as withheld:
            # guessing generously about a camera is the wrong way to be wrong.
            "allowed": ALLOWED in values,
            "raw": ",".join(values),
        })
    found.sort(key=lambda entry: entry["app"].casefold())
    return found


def snapshot() -> dict:
    if not available():
        return {
            "available": False,
            "devices": [],
            "error": "The desktop portal's permission store is not running.",
        }

    devices = []
    for device_id, label in DEVICES:
        devices.append({
            "id": device_id,
            "label": label,
            "applications": entries_for(device_id),
        })
    return {"available": True, "devices": devices, "error": ""}


def require(pattern: re.Pattern[str], value: str, message: str) -> str:
    if not pattern.match(value or ""):
        raise BoundaryError(message)
    return value


def set_permission(device: str, app: str, allowed: bool) -> None:
    require(DEVICE_ID, device, "That is not a device.")
    require(APP_ID, app, "That is not an application.")
    if not any(device == known for known, _ in DEVICES):
        raise BoundaryError("That is not a device this manages.")
    # Permissions are an array of strings, so busctl needs the element count
    # before the element -- "1 yes", not "yes".
    portal_call("SetPermission", "sbssas", TABLE, "true", device, app,
                "1", ALLOWED if allowed else DENIED)


def forget(device: str, app: str) -> None:
    """Drop the recorded answer, so the application is asked again next time."""
    require(DEVICE_ID, device, "That is not a device.")
    require(APP_ID, app, "That is not an application.")
    portal_call("DeletePermission", "sss", TABLE, device, app)


def main(arguments: list[str]) -> int:
    try:
        if arguments == ["snapshot"]:
            print(json.dumps(snapshot(), separators=(",", ":")))
            return 0

        if len(arguments) == 4 and arguments[0] == "set":
            if arguments[3] not in ("allow", "deny"):
                raise BoundaryError("That is not allow or deny.")
            set_permission(arguments[1], arguments[2], arguments[3] == "allow")
        elif len(arguments) == 3 and arguments[0] == "forget":
            forget(arguments[1], arguments[2])
        else:
            raise BoundaryError(
                "Usage: panama-permissions snapshot | set DEVICE APP allow|deny | "
                "forget DEVICE APP")
    except BoundaryError as error:
        try:
            state = snapshot()
        except BoundaryError:
            state = {"available": False, "devices": []}
        state["error"] = str(error)
        print(json.dumps(state, separators=(",", ":")))
        return 0

    print(json.dumps(snapshot(), separators=(",", ":")))
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))