From d9c9a88a66876dd567da40f5c081a92041d9f8a4 Mon Sep 17 00:00:00 2001 From: Terranom674 Date: Tue, 18 Aug 2026 08:48:11 +0200 Subject: [PATCH] Validate NC manifest SQL view identifiers --- runtime/lib/build_manifest.py | 207 ++++++++++++---------------------- 1 file changed, 71 insertions(+), 136 deletions(-) diff --git a/runtime/lib/build_manifest.py b/runtime/lib/build_manifest.py index 08c20bf..c806524 100644 --- a/runtime/lib/build_manifest.py +++ b/runtime/lib/build_manifest.py @@ -7,174 +7,109 @@ import argparse import csv import json import os +import re import subprocess import sys import tempfile from pathlib import Path, PurePosixPath +IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)?$") + + +def validate_view(name: str) -> str: + value = str(name).strip() + if not IDENTIFIER.fullmatch(value): + raise ValueError(f"invalid SQL view name: {value!r}") + return value + def read_config(path: Path) -> dict[str, tuple[str, Path]]: result: dict[str, tuple[str, Path]] = {} with path.open(encoding="utf-8") as handle: for number, line in enumerate(handle, 1): - line = line.rstrip("\n") - if not line or line.startswith("#"): - continue - fields = line.split("\t") - if len(fields) != 3: - raise ValueError(f"{path}:{number}: expected storage_id, source_prefix and local_mount") - storage_id, prefix, mount = fields - result[storage_id] = (prefix.strip("/"), Path(mount)) + line=line.rstrip("\n") + if not line or line.startswith("#"):continue + fields=line.split("\t") + if len(fields)!=3:raise ValueError(f"{path}:{number}: expected storage_id, source_prefix and local_mount") + storage_id,prefix,mount=fields + if not storage_id.strip():raise ValueError(f"{path}:{number}: storage_id is empty") + result[storage_id]=(prefix.strip("/"),Path(mount)) return result -def run_query(args: argparse.Namespace, env: dict[str, str], sql: str) -> subprocess.CompletedProcess[str]: - command = [ - "psql", "-X", "-A", "-F", "\t", "-t", - "-h", args.host, "-p", str(args.port), "-U", args.user, "-d", args.database, - "-c", sql, - ] - return subprocess.run(command, env=env, check=False, text=True, capture_output=True) +def run_query(args: argparse.Namespace, env: dict[str,str], sql: str) -> subprocess.CompletedProcess[str]: + command=["psql","-X","-A","-F","\t","-t","-v","ON_ERROR_STOP=1","-h",args.host,"-p",str(args.port),"-U",args.user,"-d",args.database,"-c",sql] + return subprocess.run(command,env=env,check=False,text=True,capture_output=True) def query_rows(args: argparse.Namespace) -> list[list[str]]: - password = args.password_file.read_text(encoding="utf-8").strip() - env = os.environ.copy() - env["PGPASSWORD"] = password - - modern_sql = f"SELECT share_id, item_type, display_name, storage_id, source_path FROM {args.view} ORDER BY share_id" - completed = run_query(args, env, modern_sql) - if completed.returncode == 0: - return list(csv.reader(completed.stdout.splitlines(), delimiter="\t")) - - # Existing installations may still expose the original four-column view. - # Keep them usable and derive folder/file from the resolved source path. - if "item_type" not in completed.stderr or "does not exist" not in completed.stderr: - raise RuntimeError(completed.stderr.strip() or "Nextcloud source view query failed") - - legacy_sql = f"SELECT share_id, display_name, storage_id, source_path FROM {args.view} ORDER BY share_id" - completed = run_query(args, env, legacy_sql) - if completed.returncode != 0: - raise RuntimeError(completed.stderr.strip() or "legacy Nextcloud source view query failed") - - rows: list[list[str]] = [] - for row in csv.reader(completed.stdout.splitlines(), delimiter="\t"): - if len(row) == 4: - share_id, display_name, storage_id, source_path = row - rows.append([share_id, "", display_name, storage_id, source_path]) - else: - rows.append(row) + password=args.password_file.read_text(encoding="utf-8").strip();env=os.environ.copy();env["PGPASSWORD"]=password;view=validate_view(args.view) + modern_sql=f"SELECT share_id, item_type, display_name, storage_id, source_path FROM {view} ORDER BY share_id" + completed=run_query(args,env,modern_sql) + if completed.returncode==0:return list(csv.reader(completed.stdout.splitlines(),delimiter="\t")) + if "item_type" not in completed.stderr or "does not exist" not in completed.stderr:raise RuntimeError(completed.stderr.strip() or "Nextcloud source view query failed") + legacy_sql=f"SELECT share_id, display_name, storage_id, source_path FROM {view} ORDER BY share_id" + completed=run_query(args,env,legacy_sql) + if completed.returncode!=0:raise RuntimeError(completed.stderr.strip() or "legacy Nextcloud source view query failed") + rows=[] + for row in csv.reader(completed.stdout.splitlines(),delimiter="\t"): + if len(row)==4: + share_id,display_name,storage_id,source_path=row;rows.append([share_id,"",display_name,storage_id,source_path]) + else:rows.append(row) return rows def contained_join(root: Path, relative: str) -> Path: - parts = PurePosixPath(relative).parts - if relative.startswith("/") or ".." in parts: - raise ValueError(f"unsafe source path: {relative}") + parts=PurePosixPath(relative).parts + if relative.startswith("/") or ".." in parts:raise ValueError(f"unsafe source path: {relative}") return root.joinpath(*parts) -def build(args: argparse.Namespace) -> dict[str, object]: - adapters = read_config(args.storage_config) - rows = query_rows(args) - if not rows and not args.allow_empty: - raise RuntimeError("Nextcloud returned no Showcase shares; refusing an empty manifest") - - manifest: list[str] = [] - errors: list[str] = [] - folder_count = 0 - file_count = 0 - +def build(args: argparse.Namespace) -> dict[str,object]: + validate_view(args.view);adapters=read_config(args.storage_config);rows=query_rows(args) + if not rows and not args.allow_empty:raise RuntimeError("Nextcloud returned no Showcase shares; refusing an empty manifest") + manifest=[];errors=[];folder_count=0;file_count=0 for row in rows: - if len(row) != 5: - errors.append(f"invalid database row with {len(row)} columns") - continue - - share_id, item_type, display_name, storage_id, source_path = row - item_type = item_type.strip().lower() - if item_type and item_type not in {"folder", "file"}: - errors.append(f"share {share_id}: unsupported item_type {item_type!r}") - continue - - adapter = adapters.get(storage_id) - if not adapter: - errors.append(f"share {share_id}: unknown storage {storage_id}") - continue - - prefix, mount = adapter - relative = source_path.strip("/") + if len(row)!=5:errors.append(f"invalid database row with {len(row)} columns");continue + share_id,item_type,display_name,storage_id,source_path=row;item_type=item_type.strip().lower() + if item_type and item_type not in {"folder","file"}:errors.append(f"share {share_id}: unsupported item_type {item_type!r}");continue + adapter=adapters.get(storage_id) + if not adapter:errors.append(f"share {share_id}: unknown storage {storage_id}");continue + prefix,mount=adapter;relative=source_path.strip("/") if prefix: - expected = prefix + "/" - if relative != prefix and not relative.startswith(expected): - errors.append(f"share {share_id}: path does not match configured prefix") - continue - relative = relative[len(prefix):].lstrip("/") - - source = contained_join(mount, relative) - if not mount.is_mount(): - errors.append(f"share {share_id}: storage mount unavailable: {mount}") - continue - + expected=prefix+"/" + if relative!=prefix and not relative.startswith(expected):errors.append(f"share {share_id}: path does not match configured prefix");continue + relative=relative[len(prefix):].lstrip("/") + source=contained_join(mount,relative) + if not mount.is_mount():errors.append(f"share {share_id}: storage mount unavailable: {mount}");continue if not item_type: - if source.is_dir(): - item_type = "folder" - elif source.is_file(): - item_type = "file" - else: - errors.append(f"share {share_id}: source unavailable: {source}") - continue - - if item_type == "folder": - if not source.is_dir(): - errors.append(f"share {share_id}: source directory unavailable: {source}") - continue - folder_count += 1 + if source.is_dir():item_type="folder" + elif source.is_file():item_type="file" + else:errors.append(f"share {share_id}: source unavailable: {source}");continue + if item_type=="folder": + if not source.is_dir():errors.append(f"share {share_id}: source directory unavailable: {source}");continue + folder_count+=1 else: - if not source.is_file(): - errors.append(f"share {share_id}: source file unavailable: {source}") - continue - file_count += 1 - + if not source.is_file():errors.append(f"share {share_id}: source file unavailable: {source}");continue + file_count+=1 + if "\t" in display_name or "\n" in display_name or "\r" in display_name:errors.append(f"share {share_id}: display name contains unsupported control characters");continue + source_text=str(source) + if "\t" in source_text or "\n" in source_text or "\r" in source_text:errors.append(f"share {share_id}: source path contains unsupported control characters");continue manifest.append(f"{share_id}\t{item_type}\t{display_name.lstrip('/')}\t{source}") - - if errors: - raise RuntimeError("; ".join(errors)) - if len(manifest) != len(rows): - raise RuntimeError("not all Showcase shares could be resolved") - - args.output.parent.mkdir(parents=True, exist_ok=True) - with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=args.output.parent, delete=False) as handle: - handle.write("\n".join(manifest) + ("\n" if manifest else "")) - temporary = Path(handle.name) + if errors:raise RuntimeError("; ".join(errors)) + if len(manifest)!=len(rows):raise RuntimeError("not all Showcase shares could be resolved") + args.output.parent.mkdir(parents=True,exist_ok=True) + with tempfile.NamedTemporaryFile("w",encoding="utf-8",dir=args.output.parent,delete=False) as handle: + handle.write("\n".join(manifest)+("\n" if manifest else ""));temporary=Path(handle.name) temporary.replace(args.output) - - return { - "shares": len(manifest), - "folders": folder_count, - "files": file_count, - "manifest": str(args.output), - } + return {"shares":len(manifest),"folders":folder_count,"files":file_count,"manifest":str(args.output)} def main() -> int: - parser = argparse.ArgumentParser() - parser.add_argument("--host", required=True) - parser.add_argument("--port", type=int, default=5432) - parser.add_argument("--database", required=True) - parser.add_argument("--user", required=True) - parser.add_argument("--password-file", required=True, type=Path) - parser.add_argument("--view", default="piwigo_showcase_sources") - parser.add_argument("--storage-config", required=True, type=Path) - parser.add_argument("--output", required=True, type=Path) - parser.add_argument("--allow-empty", action="store_true") - args = parser.parse_args() - try: - print(json.dumps(build(args), ensure_ascii=False)) - except Exception as error: - print(f"manifest: {error}", file=sys.stderr) - return 1 + parser=argparse.ArgumentParser();parser.add_argument("--host",required=True);parser.add_argument("--port",type=int,default=5432);parser.add_argument("--database",required=True);parser.add_argument("--user",required=True);parser.add_argument("--password-file",required=True,type=Path);parser.add_argument("--view",default="piwigo_showcase_sources");parser.add_argument("--storage-config",required=True,type=Path);parser.add_argument("--output",required=True,type=Path);parser.add_argument("--allow-empty",action="store_true");args=parser.parse_args() + try:print(json.dumps(build(args),ensure_ascii=False)) + except Exception as error:print(f"manifest: {error}",file=sys.stderr);return 1 return 0 -if __name__ == "__main__": - raise SystemExit(main()) +if __name__=="__main__":raise SystemExit(main())