#!/usr/bin/env python3 from __future__ import annotations import argparse import json import shutil import subprocess from pathlib import Path from typing import Any from common import DeployError, db_query_scalar, load_env_file, log DEFAULT_LIMIT = 50 ADDITIONAL_S3_AUTHORITIES_ENV = "FILE_STORAGE_ADDITIONAL_S3_AUTHORITIES_JSON" PROBLEMATIC_STATES = frozenset( { "invalid-size-bytes", "missing", "missing-runtime-s3-authority", "missing-s3-object", "missing-storage-bucket", "missing-storage-key", "noncanonical-prefixed-storage-key", "s3-access-error", "s3-authority-mismatch", "s3-size-mismatch", "size-mismatch", "unchecked-s3-provider", "unsafe-storage-key", "unsupported-provider", "unknown", } ) def sql_string_literal(value: str) -> str: return "'" + value.replace("'", "''") + "'" def split_csv_list(raw: str) -> list[str]: return [item.strip() for item in str(raw or "").split(",") if item.strip()] def unique_ordered(values: list[str]) -> list[str]: result: list[str] = [] seen: set[str] = set() for raw in values: value = str(raw or "").strip() if not value or value in seen: continue result.append(value) seen.add(value) return result def parse_size_bytes(value: Any) -> int | None: if value is None or isinstance(value, bool): return None if isinstance(value, float) and not value.is_integer(): return None try: parsed = int(value) except (TypeError, ValueError): return None return parsed if parsed >= 0 else None def normalize_s3_key_prefix(value: Any) -> str: return str(value or "").strip().strip("/") def require_runtime_s3_authority(envv: dict[str, str]) -> dict[str, str]: authority = { "provider": str(envv.get("FILE_STORAGE_PROVIDER") or "").strip().lower(), "bucket": str(envv.get("FILE_STORAGE_BUCKET") or "").strip(), "region": str(envv.get("FILE_STORAGE_REGION") or "").strip(), "key_prefix": normalize_s3_key_prefix(envv.get("FILE_STORAGE_PREFIX")), } if authority["provider"] != "s3": raise DeployError("runtime FILE_STORAGE_PROVIDER must be s3") missing = [key for key in ("bucket", "region", "key_prefix") if not authority[key]] if missing: raise DeployError( "runtime S3 authority is missing FILE_STORAGE_ values: " + ", ".join(missing) ) return authority def runtime_s3_authorities(envv: dict[str, str]) -> list[dict[str, str]]: """Return the exact runtime and source-scoped legacy S3 authorities.""" provider = str(envv.get("FILE_STORAGE_PROVIDER") or "filesystem").strip().lower() authorities: list[dict[str, str]] = [] if provider == "s3": authorities.append( { "name": "runtime", "source_type": "", **require_runtime_s3_authority(envv), } ) elif provider != "filesystem": raise DeployError(f"unsupported runtime FILE_STORAGE_PROVIDER: {provider}") raw = str(envv.get(ADDITIONAL_S3_AUTHORITIES_ENV) or "").strip() if not raw: return authorities if len(raw) >= 2 and raw[0] == raw[-1] == "'": raw = raw[1:-1] try: payload = json.loads(raw) except json.JSONDecodeError as exc: raise DeployError( f"{ADDITIONAL_S3_AUTHORITIES_ENV} must be valid JSON" ) from exc if not isinstance(payload, list): raise DeployError(f"{ADDITIONAL_S3_AUTHORITIES_ENV} must be a JSON array") seen_names = {authority["name"] for authority in authorities} seen_source_bindings: set[tuple[str, str]] = set() for index, item in enumerate(payload): if not isinstance(item, dict): raise DeployError( f"{ADDITIONAL_S3_AUTHORITIES_ENV}[{index}] must be an object" ) authority = { "name": str(item.get("name") or "").strip(), "source_type": str(item.get("source_type") or "").strip().lower(), "provider": str(item.get("provider") or "s3").strip().lower(), "bucket": str(item.get("bucket") or "").strip(), "region": str(item.get("region") or "").strip(), "key_prefix": normalize_s3_key_prefix(item.get("key_prefix")), } missing = [ key for key in ("name", "source_type", "bucket", "region", "key_prefix") if not authority[key] ] if missing: raise DeployError( f"{ADDITIONAL_S3_AUTHORITIES_ENV}[{index}] is missing: " + ", ".join(missing) ) if authority["provider"] != "s3": raise DeployError( f"{ADDITIONAL_S3_AUTHORITIES_ENV}[{index}] provider must be s3" ) if any(part in {"", ".", ".."} for part in authority["key_prefix"].split("/")): raise DeployError( f"{ADDITIONAL_S3_AUTHORITIES_ENV}[{index}] has an invalid key_prefix" ) if authority["name"] in seen_names: raise DeployError( f"{ADDITIONAL_S3_AUTHORITIES_ENV} has duplicate authority name: " f"{authority['name']}" ) binding = ( authority["source_type"], authority["bucket"], ) if binding in seen_source_bindings: raise DeployError( f"{ADDITIONAL_S3_AUTHORITIES_ENV} has an ambiguous " "source_type and bucket binding" ) seen_names.add(authority["name"]) seen_source_bindings.add(binding) authorities.append(authority) return authorities def matching_s3_authority( *, row: dict[str, Any], bucket: str, authorities: list[dict[str, str]], ) -> dict[str, str] | None: source_type = str(row.get("SourceType") or "").strip().lower() source_specific: list[dict[str, str]] = [] unscoped: list[dict[str, str]] = [] for authority in authorities: authority_source_type = str(authority.get("source_type") or "").strip().lower() if bucket != authority.get("bucket"): continue if authority_source_type: if authority_source_type == source_type: source_specific.append(authority) continue unscoped.append(authority) candidates = source_specific or unscoped if len(candidates) > 1: raise DeployError( "multiple S3 authorities match the same source_type and bucket" ) if candidates: return candidates[0] return None def validate_logical_s3_key(value: Any) -> tuple[str, str]: key = str(value or "").strip() if not key: return "", "missing-storage-key" if key.startswith("/") or "\\" in key: return key, "unsafe-storage-key" parts = key.split("/") if any(part in {"", ".", ".."} for part in parts): return key, "unsafe-storage-key" return key, "" def physical_s3_object_key(authority: dict[str, str], logical_key: str) -> str: prefix = normalize_s3_key_prefix(authority.get("key_prefix")) return f"{prefix}/{logical_key}" def require_destination_within_runtime_s3_authority( envv: dict[str, str], destination: dict[str, Any], *, context: str, ) -> dict[str, str]: runtime = require_runtime_s3_authority(envv) declared = { "provider": str(destination.get("provider") or "").strip().lower(), "bucket": str(destination.get("bucket") or "").strip(), "region": str(destination.get("region") or "").strip(), "key_prefix": normalize_s3_key_prefix(destination.get("key_prefix")), } prefix = declared["key_prefix"] within_prefix = prefix == runtime["key_prefix"] or prefix.startswith( runtime["key_prefix"] + "/" ) if ( declared["provider"] != runtime["provider"] or declared["bucket"] != runtime["bucket"] or declared["region"] != runtime["region"] or not within_prefix ): raise DeployError( f"{context} is outside the runtime FILE_STORAGE_* S3 authority" ) return declared def resolve_storage_region(envv: dict[str, str], storage_region: str = "") -> str: return ( storage_region or str(envv.get("FILE_STORAGE_REGION") or "").strip() ) def resolve_filesystem_object(root: Path, object_key: str) -> tuple[Path | None, str]: object_key = str(object_key or "").strip() if not object_key: return None, "missing-storage-key" candidate = (root / object_key).expanduser().resolve() root_resolved = root.expanduser().resolve() try: candidate.relative_to(root_resolved) except ValueError: return candidate, "unsafe-storage-key" return candidate, "" def resolve_s3_object(row: dict[str, Any]) -> tuple[str, str]: bucket = str(row.get("StorageBucket") or "").strip() key = str(row.get("StorageKey") or "").strip() return bucket, key def s3_error_state(message: str) -> str: lowered = message.lower() if "nosuchkey" in lowered or "not found" in lowered or "404" in lowered: return "missing-s3-object" return "s3-access-error" def head_s3_object( *, bucket: str, key: str, storage_region: str = "", aws_cli: str = "aws", ) -> dict[str, Any]: aws_path = shutil.which(aws_cli) if not aws_path: return {"state": "s3-access-error", "error": f"missing {aws_cli} command"} command = [ aws_path, "s3api", "head-object", "--bucket", bucket, "--key", key, "--output", "json", ] if storage_region: command.extend(["--region", storage_region]) result = subprocess.run(command, text=True, capture_output=True, check=False) if result.returncode != 0: message = (result.stderr or result.stdout or "").strip() response = {"state": s3_error_state(message)} if message: response["error"] = message[:1000] return response try: payload = json.loads(result.stdout or "{}") except json.JSONDecodeError as exc: return {"state": "s3-access-error", "error": f"invalid aws s3api head-object JSON: {exc}"} actual_size = parse_size_bytes(payload.get("ContentLength")) if actual_size is None: return { "state": "s3-access-error", "error": "aws s3api head-object response has invalid ContentLength", } response = { "state": "present", "actual_size_bytes": actual_size, } metadata = payload.get("Metadata") or {} if isinstance(metadata, dict): metadata_sha256 = str(metadata.get("sha256") or "").strip().lower() if metadata_sha256: response["metadata_sha256"] = metadata_sha256 return response def build_file_query( *, limit: int, parent_object_api_name: str = "", attachment_kind: str = "", source_type: str = "", status: str = "Active", storage_provider: str = "", parent_ids: list[str] | None = None, file_ids: list[str] | None = None, ) -> str: filters = [] if status: filters.append(f"`Status` = {sql_string_literal(status)}") if parent_object_api_name: filters.append(f"`ParentObjectApiName` = {sql_string_literal(parent_object_api_name)}") if attachment_kind: filters.append(f"`AttachmentKind` = {sql_string_literal(attachment_kind)}") if source_type: filters.append(f"`SourceType` = {sql_string_literal(source_type)}") if storage_provider: filters.append(f"`StorageProvider` = {sql_string_literal(storage_provider)}") resolved_parent_ids = unique_ordered(parent_ids or []) if resolved_parent_ids: filters.append(f"`ParentId` IN ({', '.join(sql_string_literal(item) for item in resolved_parent_ids)})") resolved_file_ids = unique_ordered(file_ids or []) if resolved_file_ids: filters.append(f"`Id` IN ({', '.join(sql_string_literal(item) for item in resolved_file_ids)})") where_clause = f"WHERE {' AND '.join(filters)}" if filters else "" limit = min(int(limit), 1000) limit_clause = f"\nLIMIT {limit}" if limit > 0 else "" return f""" SELECT JSON_OBJECT( 'Id', `Id`, 'Name', `Name`, 'OriginalFileName', `OriginalFileName`, 'ParentId', `ParentId`, 'ParentObjectApiName', `ParentObjectApiName`, 'AttachmentKind', `AttachmentKind`, 'MimeType', `MimeType`, 'FileExtension', `Extension`, 'SourceType', `SourceType`, 'SourceContentDocumentId', `SourceContentDocumentId`, 'SourceContentVersionId', `SourceContentVersionId`, 'Status', `Status`, 'StorageProvider', `StorageProvider`, 'StorageBucket', `StorageBucket`, 'StorageKey', `StorageKey`, 'StoragePath', `StoragePath`, 'Checksum', `Checksum`, 'SizeBytes', `SizeBytes`, 'CreatedDate', DATE_FORMAT(`CreatedDate`, '%Y-%m-%dT%H:%i:%sZ'), 'LastModifiedDate', DATE_FORMAT(`LastModifiedDate`, '%Y-%m-%dT%H:%i:%sZ') ) FROM `_System_File` {where_clause} ORDER BY CreatedDate DESC{limit_clause}; """.strip() def load_file_rows( envv: dict[str, str], *, limit: int, parent_object_api_name: str = "", attachment_kind: str = "", source_type: str = "", status: str = "Active", storage_provider: str = "", parent_ids: list[str] | None = None, file_ids: list[str] | None = None, ) -> list[dict[str, Any]]: sql = build_file_query( limit=limit, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, source_type=source_type, status=status, storage_provider=storage_provider, parent_ids=parent_ids, file_ids=file_ids, ) raw = db_query_scalar(envv, sql, database=envv["TIDB_DATABASE"]) rows = [] for line in raw.splitlines(): line = line.strip() if not line: continue payload = json.loads(line) if isinstance(payload, dict): rows.append(payload) return rows def check_file_row( row: dict[str, Any], *, storage_root: Path, check_s3: bool = False, storage_region: str = "", expected_s3_bucket: str = "", expected_s3_key_prefix: str = "", s3_authorities: list[dict[str, str]] | None = None, aws_cli: str = "aws", ) -> dict[str, Any]: provider = str(row.get("StorageProvider") or "").strip().lower() or "filesystem" storage_key = str(row.get("StorageKey") or row.get("StoragePath") or "").strip() expected_size = parse_size_bytes(row.get("SizeBytes")) result = { "id": str(row.get("Id") or ""), "name": str(row.get("Name") or row.get("OriginalFileName") or ""), "parent_id": str(row.get("ParentId") or ""), "parent_object_api_name": str(row.get("ParentObjectApiName") or ""), "attachment_kind": str(row.get("AttachmentKind") or ""), "source_type": str(row.get("SourceType") or ""), "status": str(row.get("Status") or ""), "storage_provider": provider, "storage_bucket": str(row.get("StorageBucket") or ""), "storage_key": storage_key, "storage_region": storage_region, "expected_size_bytes": expected_size, "state": "unknown", } if expected_size is None: result["state"] = "invalid-size-bytes" return result if provider == "s3": bucket, key = resolve_s3_object(row) result["storage_bucket"] = bucket result["storage_key"] = key if not check_s3: result["state"] = "unchecked-s3-provider" return result if not bucket: result["state"] = "missing-storage-bucket" return result logical_key, key_problem = validate_logical_s3_key(key) result["storage_key"] = logical_key if key_problem: result["state"] = key_problem return result resolved_authorities = list(s3_authorities or []) normalized_prefix = normalize_s3_key_prefix(expected_s3_key_prefix) if not resolved_authorities and expected_s3_bucket and normalized_prefix: resolved_authorities.append( { "name": "runtime", "source_type": "", "provider": "s3", "bucket": expected_s3_bucket, "region": storage_region, "key_prefix": normalized_prefix, } ) if not resolved_authorities: result["state"] = "missing-runtime-s3-authority" return result authority = matching_s3_authority( row=row, bucket=bucket, authorities=resolved_authorities, ) if authority is None: result["state"] = "s3-authority-mismatch" return result authority_region = str(authority.get("region") or "").strip() authority_prefix = normalize_s3_key_prefix(authority.get("key_prefix")) if logical_key == authority_prefix or logical_key.startswith( authority_prefix + "/" ): result["state"] = "noncanonical-prefixed-storage-key" return result object_key = physical_s3_object_key(authority, logical_key) result["storage_region"] = authority_region result["s3_authority_name"] = str(authority.get("name") or "") result["s3_object_key"] = object_key result.update( head_s3_object( bucket=bucket, key=object_key, storage_region=authority_region, aws_cli=aws_cli, ) ) if result["state"] != "present": return result actual_size = parse_size_bytes(result.get("actual_size_bytes")) if actual_size is None or actual_size != expected_size: result["state"] = "s3-size-mismatch" return result if provider != "filesystem": result["state"] = "unsupported-provider" return result object_path, problem = resolve_filesystem_object(storage_root, storage_key) if problem: result["state"] = problem if object_path is not None: result["path"] = str(object_path) return result assert object_path is not None result["path"] = str(object_path) if not object_path.is_file(): result["state"] = "missing" return result actual_size = object_path.stat().st_size result["actual_size_bytes"] = actual_size if actual_size != expected_size: result["state"] = "size-mismatch" return result result["state"] = "present" return result def summarize_results( checks: list[dict[str, Any]], *, envv: dict[str, str], limit: int, parent_object_api_name: str, attachment_kind: str, source_type: str = "", status: str = "", storage_provider: str = "", check_s3: bool = False, storage_region: str = "", s3_authorities: list[dict[str, str]] | None = None, parent_ids: list[str] | None = None, file_ids: list[str] | None = None, ) -> dict[str, Any]: counts: dict[str, int] = {} for check in checks: state = str(check.get("state") or "unknown") counts[state] = counts.get(state, 0) + 1 problems = [check for check in checks if str(check.get("state") or "unknown") != "present"] return { "status": "passed" if not problems else "failed", "env_name": envv.get("ENV_NAME", ""), "public_base_url": envv.get("PUBLIC_BASE_URL", ""), "file_storage_root": envv.get("FILE_STORAGE_ROOT", ""), "file_storage_provider": envv.get("FILE_STORAGE_PROVIDER", ""), "file_storage_bucket": envv.get("FILE_STORAGE_BUCKET", ""), "file_storage_region": storage_region, "file_storage_prefix": normalize_s3_key_prefix(envv.get("FILE_STORAGE_PREFIX")), "s3_authorities": list(s3_authorities or []), "filters": { "status": status, "parent_object_api_name": parent_object_api_name, "attachment_kind": attachment_kind, "source_type": source_type, "storage_provider": storage_provider, "check_s3": check_s3, "parent_ids": unique_ordered(parent_ids or []), "file_ids": unique_ordered(file_ids or []), "limit": limit, }, "total_checked": len(checks), "counts": dict(sorted(counts.items())), "problem_count": len(problems), "problem_samples": problems[:10], } def check_file_storage_readiness( envv: dict[str, str], *, limit: int = DEFAULT_LIMIT, parent_object_api_name: str = "", attachment_kind: str = "", source_type: str = "", status: str = "Active", storage_provider: str = "", check_s3: bool = False, storage_region: str = "", aws_cli: str = "aws", parent_ids: list[str] | None = None, file_ids: list[str] | None = None, ) -> dict[str, Any]: resolved_storage_region = resolve_storage_region(envv, storage_region) s3_authorities: list[dict[str, str]] = [] if check_s3: s3_authorities = runtime_s3_authorities(envv) runtime_authority = next( (authority for authority in s3_authorities if authority["name"] == "runtime"), None, ) if ( storage_region and runtime_authority is not None and storage_region != runtime_authority["region"] ): raise DeployError("requested S3 region is outside runtime FILE_STORAGE_REGION") if runtime_authority is not None: resolved_storage_region = runtime_authority["region"] checks = collect_file_storage_checks( envv, limit=limit, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, source_type=source_type, status=status, storage_provider=storage_provider, check_s3=check_s3, storage_region=resolved_storage_region, s3_authorities=s3_authorities, aws_cli=aws_cli, parent_ids=parent_ids, file_ids=file_ids, ) return summarize_results( checks, envv=envv, limit=limit, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, source_type=source_type, status=status, storage_provider=storage_provider, check_s3=check_s3, storage_region=resolved_storage_region, s3_authorities=s3_authorities, parent_ids=parent_ids, file_ids=file_ids, ) def collect_file_storage_checks( envv: dict[str, str], *, limit: int, parent_object_api_name: str = "", attachment_kind: str = "", source_type: str = "", status: str = "Active", storage_provider: str = "", check_s3: bool = False, storage_region: str = "", s3_authorities: list[dict[str, str]] | None = None, aws_cli: str = "aws", parent_ids: list[str] | None = None, file_ids: list[str] | None = None, ) -> list[dict[str, Any]]: storage_root_raw = str(envv.get("FILE_STORAGE_ROOT") or "").strip() if not storage_root_raw: raise DeployError("FILE_STORAGE_ROOT is required for file storage readiness checks") rows = load_file_rows( envv, limit=limit, parent_object_api_name=parent_object_api_name, attachment_kind=attachment_kind, source_type=source_type, status=status, storage_provider=storage_provider, parent_ids=parent_ids, file_ids=file_ids, ) storage_root = Path(storage_root_raw).expanduser().resolve() resolved_storage_region = resolve_storage_region(envv, storage_region) resolved_s3_authorities = list(s3_authorities or []) if check_s3: if not resolved_s3_authorities: resolved_s3_authorities = runtime_s3_authorities(envv) runtime_authority = next( ( authority for authority in resolved_s3_authorities if authority["name"] == "runtime" ), None, ) if ( storage_region and runtime_authority is not None and storage_region != runtime_authority["region"] ): raise DeployError("requested S3 region is outside runtime FILE_STORAGE_REGION") if runtime_authority is not None: resolved_storage_region = runtime_authority["region"] return [ check_file_row( row, storage_root=storage_root, check_s3=check_s3, storage_region=resolved_storage_region, s3_authorities=resolved_s3_authorities, aws_cli=aws_cli, ) for row in rows ] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Check that _System_File metadata has reachable storage objects.") parser.add_argument("--env-file", required=True) parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT) parser.add_argument("--parent-object-api-name", default="") parser.add_argument("--attachment-kind", default="") parser.add_argument("--source-type", default="") parser.add_argument("--status", default="Active") parser.add_argument("--storage-provider", default="") parser.add_argument("--parent-id", action="append", default=[]) parser.add_argument("--file-id", action="append", default=[]) parser.add_argument("--check-s3", action="store_true") parser.add_argument("--storage-region", default="") parser.add_argument("--aws-cli", default="aws") parser.add_argument("--summary-json") parser.add_argument("--fail-on-problems", action="store_true") parser.add_argument("--fail-on-empty", action="store_true") return parser.parse_args() def main() -> int: args = parse_args() envv = load_env_file(Path(args.env_file).expanduser().resolve()) summary = check_file_storage_readiness( envv, limit=args.limit, parent_object_api_name=args.parent_object_api_name, attachment_kind=args.attachment_kind, source_type=args.source_type, status=args.status, storage_provider=args.storage_provider, check_s3=args.check_s3, storage_region=args.storage_region, aws_cli=args.aws_cli, parent_ids=[item for raw in args.parent_id for item in split_csv_list(raw)], file_ids=[item for raw in args.file_id for item in split_csv_list(raw)], ) output = json.dumps(summary, indent=2, ensure_ascii=False) print(output) if args.summary_json: summary_path = Path(args.summary_json).expanduser().resolve() summary_path.parent.mkdir(parents=True, exist_ok=True) summary_path.write_text(output + "\n", encoding="utf-8") log(f"wrote file storage readiness summary: {summary_path}") if args.fail_on_problems and summary["problem_count"] > 0: return 1 if args.fail_on_empty and summary["total_checked"] == 0: log("file storage readiness checked 0 files") return 1 return 0 if __name__ == "__main__": raise SystemExit(main())