pigway-frame-flow/pigway_frame_flow/providers/synology_photos.py

"""Optional, read-only Synology Photos person catalog with pinned TLS."""

import hashlib
import http.client
import json
import os
import ssl
import threading
import time
from urllib.parse import urlencode, urlsplit

from pydantic import Field, SecretStr

from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.models import StrictModel


class PhotosLogin(StrictModel):
    username: str = Field(min_length=1, max_length=120)
    password: SecretStr = Field(min_length=1, max_length=256)


class PhotosConnection(StrictModel):
    configured: bool
    connected: bool


class PhotosPerson(StrictModel):
    id: int
    name: str
    count: int = 0
    named: bool = False


class SynologyPhotos:
    def __init__(self, settings):
        self.settings = settings
        self.path = settings.config_dir / "synology-photos-session.json"
        self.lock = threading.RLock()
        self.cache = {}
        if self.path.is_file():
            try:
                self.sid()
            except FlowError:
                pass

    def endpoint(self):
        url = urlsplit(os.getenv("SYNOLOGY_PHOTOS_URL", ""))
        pin = os.getenv("SYNOLOGY_PHOTOS_CERT_SHA256", "").lower().replace(":", "")
        if (
            url.scheme != "https"
            or not url.hostname
            or url.username
            or url.password
            or url.query
            or url.fragment
            or url.path not in ("", "/")
            or len(pin) != 64
            or any(c not in "0123456789abcdef" for c in pin)
        ):
            raise FlowError(
                E.CONFIG_REQUIRED,
                "Configure Synology Photos URL and certificate fingerprint",
                503,
            )
        return url, pin

    def status(self):
        try:
            self.endpoint()
            configured = True
        except FlowError:
            configured = False
        return {
            "configured": configured,
            "connected": configured and self.path.is_file(),
        }

    def request(self, api, method, version, **params):
        # No arbitrary RPC forwarding or mutation methods, even for administrators.
        if (api, method) not in {
            ("SYNO.API.Auth", "login"),
            ("SYNO.API.Auth", "logout"),
            ("SYNO.FotoTeam.Browse.Person", "list"),
            ("SYNO.FotoTeam.Browse.Item", "list"),
            ("SYNO.FotoTeam.Browse.Folder", "get"),
        }:
            raise FlowError(E.FORBIDDEN, "Unsupported Photos query", 403)
        url, pin = self.endpoint()
        connection = http.client.HTTPSConnection(
            url.hostname,
            url.port or 443,
            timeout=30,
            context=ssl._create_unverified_context(),
        )
        try:
            connection.connect()
            certificate = connection.sock.getpeercert(binary_form=True)
            if not certificate or hashlib.sha256(certificate).hexdigest() != pin:
                raise FlowError(E.FORBIDDEN, "Synology Photos certificate changed", 403)
            # Check the certificate on this exact socket BEFORE transmitting a secret.
            data = urlencode(
                {"api": api, "method": method, "version": version, **params}
            )
            connection.request(
                "POST",
                "/webapi/entry.cgi",
                data,
                {"Content-Type": "application/x-www-form-urlencoded"},
            )
            response = connection.getresponse()
            raw = response.read(4 * 1024 * 1024 + 1)
            if response.status != 200 or len(raw) > 4 * 1024 * 1024:
                raise ValueError()
            result = json.loads(raw)
            if not result.get("success"):
                code = result.get("error", {}).get("code")
                if api == "SYNO.API.Auth" or code in (105, 106, 107, 119):
                    if api != "SYNO.API.Auth":
                        self.disconnect()
                    raise FlowError(
                        E.CONFIG_REQUIRED, "Reconnect Synology Photos account", 409
                    )
                raise FlowError(
                    E.CLOUD_UNAVAILABLE, "Synology Photos query was rejected", 502
                )
            return result.get("data", {})
        except (OSError, ValueError, http.client.HTTPException):
            raise FlowError(
                E.CLOUD_UNAVAILABLE, "Synology Photos is unavailable", 502
            ) from None
        finally:
            connection.close()

    def sid(self):
        try:
            sid = json.loads(self.path.read_text())["sid"]
            if not isinstance(sid, str) or not sid:
                raise ValueError()
        except (OSError, ValueError, KeyError):
            raise FlowError(
                E.CONFIG_REQUIRED, "Connect Synology Photos account first", 409
            ) from None
        if sid not in self.settings.additional_secrets:
            self.settings.additional_secrets.append(sid)
        return sid

    def connect(self, payload):
        password = payload.password.get_secret_value()
        self.settings.additional_secrets.append(password)
        with self.lock:
            sid = self.request(
                "SYNO.API.Auth",
                "login",
                7,
                account=payload.username,
                passwd=password,
                session="SynologyPhotos",
                format="sid",
            )["sid"]
            self.settings.additional_secrets.append(sid)
            self.path.parent.mkdir(parents=True, exist_ok=True)
            temporary = self.path.with_suffix(".tmp")
            fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
            with os.fdopen(fd, "w") as stream:
                json.dump({"sid": sid}, stream)
            os.chmod(temporary, 0o600)
            temporary.replace(self.path)
            self.cache.clear()
        return self.status()

    def disconnect(self):
        with self.lock:
            self.path.unlink(missing_ok=True)
            self.cache.clear()
        return self.status()

    def pages(self, api, version, **params):
        rows = []
        for offset in range(0, 100000, 500):
            page = self.request(
                api,
                "list",
                version,
                offset=offset,
                limit=500,
                _sid=self.sid(),
                **params,
            ).get("list", [])
            rows.extend(page)
            if len(page) < 500:
                return rows
        raise FlowError(E.CONFLICT, "Photos catalog exceeds query limit", 409)

    def people(self):
        with self.lock:
            rows = self.pages("SYNO.FotoTeam.Browse.Person", 3)
            return [
                {
                    "id": r["id"],
                    "name": r.get("name") or str(r["id"]),
                    "count": r.get("item_count", 0),
                    "named": bool(r.get("name")),
                }
                for r in rows
            ]

    def paths(self, person_ids, mode, refresh=False):
        key = (tuple(sorted(person_ids)), mode)
        with self.lock:
            cached = self.cache.get(key)
            if cached and not refresh and time.monotonic() - cached[0] < 900:
                return cached[1]
            known = {p["id"] for p in self.people()}
            ids = person_ids or sorted(known)
            if not set(ids) <= known:
                raise FlowError(E.INVALID_REQUEST, "Selected person no longer exists")
            sets = []
            folders = {}
            for person in ids:
                paths = set()
                for row in self.pages(
                    "SYNO.FotoTeam.Browse.Item",
                    7,
                    person_id=person,
                    additional=json.dumps(["folder"]),
                ):
                    folder = row.get("additional", {}).get("folder", {})
                    if isinstance(folder, str):
                        name = folder
                    elif isinstance(folder, dict) and folder.get("name"):
                        name = folder["name"]
                    else:
                        folder_id = row.get("folder_id")
                        if folder_id is None:
                            raise FlowError(
                                E.CLOUD_UNAVAILABLE,
                                "Photos response has no folder identity",
                                502,
                            )
                        if folder_id not in folders:
                            folders[folder_id] = self.request(
                                "SYNO.FotoTeam.Browse.Folder",
                                "get",
                                2,
                                id=folder_id,
                                _sid=self.sid(),
                            )["folder"]["name"]
                        name = folders[folder_id]
                    if "filename" not in row:
                        raise FlowError(
                            E.CLOUD_UNAVAILABLE, "Photos response has no filename", 502
                        )
                    paths.add(name.strip("/") + "/" + row["filename"])
                sets.append(paths)
            selected = (
                (set.intersection(*sets) if mode == "all" else set.union(*sets))
                if sets
                else set()
            )
            self.cache[key] = (time.monotonic(), selected)
            return selected