import sys sys.dont_write_bytecode = True import argparse import fnmatch import hashlib import os import shutil from db_backend import database_url_from_env, is_postgres_url from paths import apply_path_config, default_project_paths from migrate_runtime_safety import require_runtime_hardening_stopped from postgres_runtime import load_postgres_environment from runtime_security import ClusterAuthorityLock, reject_reparse_components DESKTOP_HF = r"C:\Users\pro100noob\Desktop\HugginFace" EXCLUDED_DIRS = {"__pycache__", ".git", ".opencode", "node_modules", "tmp", "runtime"} APP_FILES = [ "app.py", "console_runner.py", "dashboard.py", "keycheck_runner.py", "migrate_layout.py", "paths.py", "scan_manager.py", "scanner.py", "scanner_db.py", "supervisor.py", "ui_components.py", "config.yaml", "secrets.yaml", "requirements.txt", "CHEATSHEET.md", "DETECTOR_NOTES.md", ] APP_DIRS = [".streamlit", "keycheckers"] RESULT_FILES = ["found_secrets.jsonl", "scan_results.jsonl", "scan_errors.log", "scanner.db", "scanner.db-wal", "scanner.db-shm"] KEYCHECK_SERVICE_SCRIPTS = { "anthropic": [os.path.join("anthropic", "anthropicKeycheck.py")], "aws": [os.path.join("aws", "awsKeycheck.py")], "azure": [os.path.join("azure", "azureKeycheck.py")], "deepseek": [os.path.join("deepseek", "deepseekKeycheck.py")], "dockerhub": [os.path.join("dockerhub", "dockerhubKeycheck.py"), os.path.join("dockerhub", "dockerhub.txt")], "gcp": [os.path.join("gcp", "gcpKeycheck.py"), os.path.join("gcp", "gcp.txt")], "gemini": [os.path.join("gemini", "geminiKeycheck.py"), os.path.join("gemini", "gem.txt")], "groq": [os.path.join("groq", "groqKeycheck.py")], "github": [os.path.join("github", "githubKeycheck.py"), os.path.join("github", "github.txt")], "gitlab": [os.path.join("gitlab", "gitlabKeycheck.py"), os.path.join("gitlab", "gitlab.txt")], "kimi": [os.path.join("kimi", "kimiKeycheck.py")], "openai": ["Keycheck.py"], "openrouter": ["OpenrouterKeycheck.py"], "provider_resolver": [os.path.join("provider_resolver", "providerResolverKeycheck.py")], "qwen": [os.path.join("qwen", "qwenKeycheck.py")], "replicate": [os.path.join("replicate", "replicateKeycheck.py")], "xai": [os.path.join("xai", "xaiKeycheck.py")], "huggingface": [os.path.join("huggingface", "huggingfaceKeycheck.py")], "zai": [os.path.join("zai", "zaiKeycheck.py")], } KEYCHECK_TOP_LEVEL_PREFIXES = { "openai": "openai", "openrouter": "openrouter", } def load_config(config_path): if not config_path: return {'global': default_project_paths()} try: import yaml except ImportError as e: raise SystemExit("PyYAML is required for --config") from e with open(config_path, "r", encoding="utf-8") as f: config = apply_path_config(yaml.safe_load(f) or {}, config_path) return config def load_layout(config_path): return load_config(config_path).get('global') or {} def mkdir(path, dry_run=False): if dry_run: print(f"mkdir {path}") return os.makedirs(path, exist_ok=True) def copy_file(src, dst, overwrite=False, dry_run=False): if not os.path.exists(src): return False if os.path.lexists(dst) and not overwrite: print(f"skip existing {dst}") return False parent = os.path.dirname(dst) if parent: mkdir(parent, dry_run) if dry_run: print(f"copy {src} -> {dst}") return True try: reject_reparse_components(parent or os.path.dirname(os.path.abspath(dst))) if os.path.lexists(dst): reject_reparse_components(dst) shutil.copy2(src, dst) except OSError as e: print(f"skip locked/unavailable {src}: {e}") return False print(f"copied {src} -> {dst}") return True def ignore_app_dir(_dir, names): ignored = set() for name in names: if name in EXCLUDED_DIRS: ignored.add(name) if fnmatch.fnmatch(name, "*.pyc"): ignored.add(name) return ignored def copy_dir(src, dst, overwrite=False, dry_run=False, verified_apply=False): if not os.path.isdir(src): return False if os.path.lexists(dst) and not overwrite: print(f"skip existing {dst}") return False if dry_run: print(f"copytree {src} -> {dst}") return True if os.path.lexists(dst) and overwrite: raise RuntimeError('legacy directory replacement is retired; existing directories are never replaced') shutil.copytree(src, dst, ignore=ignore_app_dir, dirs_exist_ok=False) print(f"copied {src} -> {dst}") return True def create_layout(layout, dry_run=False): for key in ("project_dir", "runtime_dir", "results_dir", "queue_dir", "log_dir", "state_dir", "keycheck_dir", "work_dir"): mkdir(layout[key], dry_run) mkdir(os.path.join(layout["runtime_dir"], "imports"), dry_run) def copy_app_files(source_dir, layout, overwrite=False, dry_run=False, verified_apply=False): project_dir = layout["project_dir"] if os.path.abspath(source_dir) == os.path.abspath(project_dir): print("app source is already project_dir; app copy skipped") return for name in APP_FILES: copy_file(os.path.join(source_dir, name), os.path.join(project_dir, name), overwrite, dry_run) for name in APP_DIRS: copy_dir(os.path.join(source_dir, name), os.path.join(project_dir, name), overwrite, dry_run, verified_apply) def copy_scanner_runtime(old_root, layout, overwrite=False, dry_run=False, verified_apply=False): for name in RESULT_FILES: copy_file(os.path.join(old_root, name), os.path.join(layout["results_dir"], name), overwrite, dry_run) for pattern in ("todo_*.txt", "checked_*.txt"): if not os.path.isdir(old_root): continue for name in os.listdir(old_root): if fnmatch.fnmatch(name, pattern): copy_file(os.path.join(old_root, name), os.path.join(layout["queue_dir"], name), overwrite, dry_run) copy_dir(os.path.join(old_root, "logs"), layout["log_dir"], overwrite, dry_run, verified_apply) copy_dir(os.path.join(old_root, "state"), layout["state_dir"], overwrite, dry_run, verified_apply) copy_file(os.path.join(old_root, "runner_state.json"), os.path.join(layout["state_dir"], "runner_state.json"), overwrite, dry_run) def line_hash(line): return hashlib.sha256(line.strip().encode("utf-8", errors="replace")).hexdigest() def existing_line_hashes(path): hashes = set() if not os.path.exists(path): return hashes with open(path, "r", encoding="utf-8", errors="replace") as f: for line in f: if line.strip(): hashes.add(line_hash(line)) return hashes def import_jsonl_dedupe(inputs, output, dry_run=False): hashes = existing_line_hashes(output) added = 0 if dry_run: print(f"dedupe import {len(inputs)} file(s) -> {output}") return 0 mkdir(os.path.dirname(output), dry_run=False) with open(output, "a", encoding="utf-8") as dst: for path in inputs: if not os.path.exists(path): continue with open(path, "r", encoding="utf-8", errors="replace") as src: for line in src: if not line.strip(): continue digest = line_hash(line) if digest in hashes: continue dst.write(line if line.endswith("\n") else line + "\n") hashes.add(digest) added += 1 print(f"imported {added} unique finding line(s) into {output}") return added def copy_legacy_keychecker_outputs(layout, desktop_dir=DESKTOP_HF, overwrite=False, dry_run=False): if not os.path.isdir(desktop_dir): return for service in KEYCHECK_SERVICE_SCRIPTS: source_dir = os.path.join(desktop_dir, service) target_dir = os.path.join(layout["keycheck_dir"], service) if os.path.isdir(source_dir): for name in os.listdir(source_dir): if name.lower().endswith((".txt", ".jsonl")): copy_file(os.path.join(source_dir, name), os.path.join(target_dir, name), overwrite, dry_run) for service, prefix in KEYCHECK_TOP_LEVEL_PREFIXES.items(): target_dir = os.path.join(layout["keycheck_dir"], service) for name in os.listdir(desktop_dir): lower = name.lower() if lower.startswith(prefix) and lower.endswith((".txt", ".jsonl")): copy_file(os.path.join(desktop_dir, name), os.path.join(target_dir, name), overwrite, dry_run) def merge_unique_lines(inputs, output, dry_run=False): values = [] seen = existing_values = set() if os.path.exists(output): with open(output, "r", encoding="utf-8", errors="replace") as f: existing_values = {line.strip().lstrip("\ufeff") for line in f if line.strip()} seen = set(existing_values) for path in inputs: if not os.path.exists(path): continue with open(path, "r", encoding="utf-8", errors="replace") as f: for line in f: value = line.strip().lstrip("\ufeff") if value and value not in seen: values.append(value) seen.add(value) if dry_run: print(f"merge {len(values)} unique line(s) -> {output}") return mkdir(os.path.dirname(output), dry_run=False) with open(output, "a", encoding="utf-8") as f: for value in values: f.write(value + "\n") print(f"appended {len(values)} unique line(s) -> {output}") def import_huggingface_desktop(layout, desktop_dir=DESKTOP_HF, overwrite=False, dry_run=False): if not os.path.isdir(desktop_dir): print(f"Desktop HugginFace directory not found: {desktop_dir}") return jsonl_inputs = [os.path.join(desktop_dir, name) for name in os.listdir(desktop_dir) if fnmatch.fnmatch(name, "found_secrets*.jsonl")] import_jsonl_dedupe(jsonl_inputs, os.path.join(layout["results_dir"], "found_secrets.jsonl"), dry_run) merge_unique_lines([os.path.join(desktop_dir, "checked.txt")], os.path.join(layout["queue_dir"], "checked_huggingface.txt"), dry_run) merge_unique_lines([os.path.join(desktop_dir, "todo.txt")], os.path.join(layout["queue_dir"], "todo_huggingface.txt"), dry_run) copy_file(os.path.join(desktop_dir, "proxy.txt"), layout["proxy_file"], overwrite=False, dry_run=dry_run) copy_file(os.path.join(desktop_dir, "requirements-keycheckers.txt"), os.path.join(layout["project_dir"], "requirements-keycheckers.txt"), overwrite, dry_run) copy_file(os.path.join(desktop_dir, "KEYCHECKERS.md"), os.path.join(layout["project_dir"], "KEYCHECKERS.md"), overwrite, dry_run) for service, rel_paths in KEYCHECK_SERVICE_SCRIPTS.items(): for rel_path in rel_paths: src = os.path.join(desktop_dir, rel_path) dst = os.path.join(layout["project_dir"], "keycheckers", service, os.path.basename(rel_path)) copy_file(src, dst, overwrite, dry_run) copy_legacy_keychecker_outputs(layout, desktop_dir, overwrite, dry_run) def parse_args(): parser = argparse.ArgumentParser(description="Copy/import legacy scanner files into the unified D:\\truf layout.") parser.add_argument("--config", default="config.yaml") parser.add_argument("--source-app", default=os.path.dirname(os.path.abspath(__file__))) parser.add_argument("--target-app", help="Destination app directory. Defaults to \\app.") parser.add_argument("--in-place", action="store_true", help="Use project_dir from config instead of copying to \\app.") parser.add_argument("--old-root", default=r"D:\truf") parser.add_argument("--desktop-hf", default=DESKTOP_HF) parser.add_argument("--overwrite", action="store_true") parser.add_argument("--dry-run", action="store_true") parser.add_argument("--apply", action="store_true", help="Retired; production layout mutation is disabled") parser.add_argument("--no-app-copy", action="store_true") parser.add_argument("--no-desktop-import", action="store_true") return parser.parse_args() def main(): args = parse_args() if args.apply and args.dry_run: raise SystemExit('--apply and --dry-run are mutually exclusive') if args.apply: raise SystemExit( 'migrate_layout --apply is retired because the production layout is already migrated. ' 'Use reviewed offline backup/restore tooling for any future relocation.' ) config = load_config(args.config) layout = config.get('global') or {} if not args.in_place: layout["project_dir"] = args.target_app or os.path.join(layout["root_dir"], "app") dry_run = not args.apply load_postgres_environment(os.path.abspath(args.config), config) endpoint_dsn = database_url_from_env() or layout.get('database_url') if not is_postgres_url(endpoint_dsn): raise SystemExit('A caller-selected canonical PostgreSQL DSN is required for maintenance authority') with ClusterAuthorityLock(config, endpoint_dsn=endpoint_dsn): require_runtime_hardening_stopped(config) create_layout(layout, dry_run) if not args.no_app_copy: copy_app_files(args.source_app, layout, args.overwrite, dry_run, verified_apply=args.apply) copy_scanner_runtime(args.old_root, layout, args.overwrite, dry_run, verified_apply=args.apply) if not args.no_desktop_import: import_huggingface_desktop(layout, args.desktop_hf, args.overwrite, dry_run) if dry_run: print("Dry-run complete. --apply is retired; use reviewed offline backup/restore tooling for relocation.") return 0 print("Migration copy/import finished. Originals were left in place.") return 0 if __name__ == "__main__": main()