mirror of
https://github.com/Terranom674/Piwigo_Bratonien_Tools.git
synced 2026-09-19 14:04:34 +00:00
Validate NC manifest SQL view identifiers
This commit is contained in:
Terranom674
@@ -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())
|
||||
|
||||
Reference in New Issue
Block a user