Files
pigway-frame-flow/pigway_frame_flow/providers/photo_sources.py
T

369 lines
14 KiB
Python

"""Read-only image catalogs; provider credentials never enter SQLite."""
import base64
import json
import os
import posixpath
import random
import stat
import time
from pathlib import Path
from typing import Literal
from urllib.parse import unquote, urljoin, urlsplit
from xml.etree import ElementTree
from pydantic import Field
from pigway_frame_flow.errors import ErrorCode as E
from pigway_frame_flow.errors import FlowError
from pigway_frame_flow.models import StrictModel
from pigway_frame_flow.providers.photo_names import path_caption
from pigway_frame_flow.settings import secret
from pigway_frame_flow.storage.database import encode
class PhotoSourceInput(StrictModel):
name: str = Field(min_length=1, max_length=120)
kind: Literal["local", "webdav"]
location: str = Field(min_length=1, max_length=2048)
recursive: bool = True
enabled: bool = True
catalog_limit: int = Field(default=10000, ge=1, le=100000)
organized_only: bool = False
folder_caption: bool = False
credential_env: str | None = Field(
default=None, pattern=r"^FRAME_FLOW_SOURCE_[A-Z0-9_]+$"
)
class PhotoSourceView(PhotoSourceInput):
id: str
class PhotoItem(StrictModel):
id: str
name: str
caption: dict[str, str] | None = None
class PhotoCatalog(StrictModel):
items: list[PhotoItem]
total: int
offset: int
limit: int
catalog_limit: int
class PhotoSources:
extensions = {".jpg", ".jpeg", ".png", ".webp"}
def __init__(self, db, settings, assets):
self.db, self.settings, self.assets = db, settings, assets
self.cache = {}
# Register provider secrets before request IDs or errors can be reflected.
for row in self.list():
if row["kind"] == "webdav" and row.get("credential_env"):
try:
self.headers(
PhotoSourceInput.model_validate(
{k: v for k, v in row.items() if k != "id"}
)
)
except FlowError:
pass
def list(self):
return [
dict(json.loads(row["payload"]), id=row["key"][13:])
for row in self.db.all(
"SELECT key,payload FROM app_config WHERE key LIKE 'photo_source:%' ORDER BY key"
)
]
def get(self, source_id):
row = self.db.one(
"SELECT payload FROM app_config WHERE key=?", ("photo_source:" + source_id,)
)
if not row:
raise FlowError(E.NOT_FOUND, "Photo source not found", 404)
return PhotoSourceInput.model_validate_json(row["payload"])
def save(self, source_id, source):
if source.kind == "local":
self.local_directory(source.location)
if source.credential_env:
raise FlowError(
E.INVALID_REQUEST, "Local sources do not use credentials"
)
else:
source = source.model_copy(
update={
"recursive": False,
"location": source.location.rstrip("/") + "/",
}
)
parsed = urlsplit(source.location)
if (
parsed.scheme != "https"
or not parsed.hostname
or parsed.username
or parsed.password
or parsed.query
or parsed.fragment
or parsed.port not in (None, 443)
):
raise FlowError(
E.INVALID_REQUEST, "WebDAV requires a plain HTTPS directory URL"
)
if parsed.hostname.lower() not in self.settings.image_url_hosts:
raise FlowError(
E.FORBIDDEN, "Configure the allowed image host first", 403
)
self.db.execute(
"INSERT OR REPLACE INTO app_config(key,payload) VALUES(?,?)",
("photo_source:" + source_id, encode(source.model_dump())),
)
self.cache.pop(source_id, None)
return dict(source.model_dump(), id=source_id)
def delete(self, source_id):
self.get(source_id)
for row in self.db.all("SELECT payload FROM jobs"):
if json.loads(row["payload"])["source"].get("source_id") == source_id:
raise FlowError(E.CONFLICT, "Remove jobs using this source first", 409)
for row in self.db.all(
"SELECT payload FROM operations WHERE status IN ('QUEUED','RENDERING','UPLOADING','SYNCING','UNKNOWN')"
):
payload = json.loads(row["payload"])
if payload.get("content", payload).get("source_id") == source_id:
raise FlowError(E.CONFLICT, "This source has pending operations", 409)
self.db.execute(
"DELETE FROM app_config WHERE key=?", ("photo_source:" + source_id,)
)
self.cache.pop(source_id, None)
def local_directory(self, location):
root = self.settings.photo_root.resolve()
directory = (root / location).resolve()
if Path(location).is_absolute() or not directory.is_relative_to(root):
raise FlowError(
E.FORBIDDEN,
"Photo directory must be inside the mounted photo root",
403,
)
if not directory.is_dir():
raise FlowError(
E.NOT_FOUND,
"Photo directory is unavailable; check its read-only mount",
404,
)
try:
with os.scandir(directory):
pass
except PermissionError:
raise FlowError(
E.FORBIDDEN,
"Photo directory is not readable; configure the container's existing NAS read group",
403,
) from None
return directory
def headers(self, source):
if not source.credential_env:
return {}
try:
value = secret(source.credential_env)
except OSError:
raise FlowError(
E.CONFIG_REQUIRED, "WebDAV credential secret is unavailable"
) from None
if value and value not in self.settings.additional_secrets:
self.settings.additional_secrets.append(value)
try:
credentials = json.loads(value)
username, password = credentials["username"], credentials["password"]
if (
not isinstance(username, str)
or not isinstance(password, str)
or ":" in username
):
raise ValueError()
except (ValueError, KeyError, TypeError):
raise FlowError(
E.CONFIG_REQUIRED,
"WebDAV credential secret must contain username and password",
) from None
for sensitive in (value, username, password):
if sensitive and sensitive not in self.settings.additional_secrets:
self.settings.additional_secrets.append(sensitive)
encoded = base64.b64encode((username + ":" + password).encode()).decode()
if encoded not in self.settings.additional_secrets:
self.settings.additional_secrets.append(encoded)
return {"Authorization": "Basic " + encoded}
def catalog(self, source_id, refresh=False):
source = self.get(source_id)
if not source.enabled:
raise FlowError(E.CONFLICT, "Photo source is disabled", 409)
cached = self.cache.get(source_id)
if cached and not refresh and time.monotonic() - cached[0] < 60:
return cached[1]
items = []
if source.kind == "local":
directory = self.local_directory(source.location)
root = self.settings.photo_root.resolve()
# Do not follow symlink directories or import Synology's thumbnail trees.
for folder, dirs, files in os.walk(directory, followlinks=False):
dirs[:] = (
sorted(
d
for d in dirs
if d != "#recycle"
and not d.startswith(".")
and not d.startswith("@")
and not Path(folder, d).is_symlink()
)
if source.recursive
else []
)
caption = path_caption(
Path(folder).relative_to(root).as_posix() + "/__photo__.jpg"
)
if source.organized_only and caption is None:
continue
for name in sorted(files):
path = Path(folder, name)
if (
name.startswith(".")
or path.suffix.lower() not in self.extensions
or path.is_symlink()
):
continue
try:
if path.stat().st_size > self.settings.max_bytes:
continue
relative = path.relative_to(root).as_posix()
items.append(
{
"id": relative,
"name": name,
"caption": caption if source.folder_caption else None,
}
)
except OSError:
continue
if len(items) >= source.catalog_limit:
break
if len(items) >= source.catalog_limit:
break
else:
data = self.assets.fetch_url(
source.location,
method="PROPFIND",
headers={**self.headers(source), "Depth": "1"},
limit=2 * 1024 * 1024,
accepted_status=(207,),
)
if (
b"\x00" in data
or b"<!DOCTYPE" in data.upper()
or b"<!ENTITY" in data.upper()
):
raise FlowError(E.INVALID_REQUEST, "Unsafe WebDAV XML")
try:
document = ElementTree.fromstring(data)
base = urlsplit(source.location)
prefix = posixpath.normpath(unquote(base.path)).rstrip("/") + "/"
seen = set()
for response in document.findall("{DAV:}response"):
href = response.findtext("{DAV:}href")
if not href or response.find(".//{DAV:}collection") is not None:
continue
url = urljoin(source.location, href)
parsed = urlsplit(url)
if (
(parsed.scheme, parsed.netloc) != (base.scheme, base.netloc)
or not posixpath.normpath(unquote(parsed.path)).startswith(
prefix
)
or "\\" in unquote(parsed.path)
or url in seen
or parsed.query
or parsed.fragment
):
continue
caption = path_caption(unquote(parsed.path))
if source.organized_only and caption is None:
continue
if Path(parsed.path).suffix.lower() in self.extensions:
seen.add(url)
items.append(
{
"id": url,
"name": unquote(Path(parsed.path).name),
"caption": caption if source.folder_caption else None,
}
)
if len(items) >= source.catalog_limit:
break
except ElementTree.ParseError:
raise FlowError(E.INVALID_REQUEST, "Invalid WebDAV catalog") from None
self.cache[source_id] = (time.monotonic(), items)
return items
def read(self, source_id, item_id=None):
source = self.get(source_id)
items = self.catalog(source_id)
if not items:
raise FlowError(
E.NOT_FOUND, "Photo source contains no supported images", 404
)
item = (
next((x for x in items if x["id"] == item_id), None)
if item_id
else random.SystemRandom().choice(items)
)
if item is None:
raise FlowError(E.NOT_FOUND, "Photo is not in this source", 404)
if source.kind == "local":
root = self.settings.photo_root.resolve()
path = root / item["id"]
if path.is_symlink() or not path.resolve().is_relative_to(
self.local_directory(source.location)
):
raise FlowError(E.FORBIDDEN, "Photo path is outside its source", 403)
try:
# Open every component without following links; a renamed parent
# must not turn a previously cataloged image into a secret-file read.
parent_fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY)
try:
components = path.relative_to(root).parts
for component in components[:-1]:
child_fd = os.open(
component,
os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW,
dir_fd=parent_fd,
)
os.close(parent_fd)
parent_fd = child_fd
file_fd = os.open(
components[-1],
os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK,
dir_fd=parent_fd,
)
with os.fdopen(file_fd, "rb") as handle:
if not stat.S_ISREG(os.fstat(handle.fileno()).st_mode):
raise FlowError(
E.FORBIDDEN, "Photo must be a regular file", 403
)
data = handle.read(self.settings.max_bytes + 1)
finally:
os.close(parent_fd)
except OSError:
raise FlowError(E.NOT_FOUND, "Photo is unavailable", 404) from None
else:
data = self.assets.fetch_url(item["id"], headers=self.headers(source))
if len(data) > self.settings.max_bytes:
raise FlowError(E.INVALID_REQUEST, "Image exceeds size limit")
return data, item["name"], item.get("caption")