import sys sys.dont_write_bytecode = True if not sys.dont_write_bytecode: raise RuntimeError('dashboard could not disable bytecode writes') import argparse from collections import Counter from datetime import datetime, timedelta, timezone import hashlib import html import ipaddress import json import os import re import shutil import sqlite3 import time import pandas as pd import plotly.express as px import streamlit as st from scanner_db import DB_FILENAME, get_database_path, queue_counts, sanitize_endpoint from paths import apply_path_config, default_project_paths, resolve_optional_path from db_backend import connect_postgres, connect_sqlite, database_url_from_env, is_postgres_url from lifecycle_authority import CHILD_KIND_ENV, LifecycleAuthorityError, require_active_supervisor_child DEFAULT_PATHS = default_project_paths() DEFAULT_RESULTS_DIR = os.getenv('SCAN_RESULTS_DIR', DEFAULT_PATHS['results_dir']) DEFAULT_QUEUE_DIR = DEFAULT_PATHS['queue_dir'] DEFAULT_STATE_FILE = DEFAULT_PATHS['state_file'] DEFAULT_LOG_DIR = DEFAULT_PATHS['log_dir'] SOURCES = ['github', 'github_archive', 'github_archive_files', 'github_gists', 'gitlab', 'github_actions', 'gitlab_ci', 'docker', 'dockerhub', 'npm', 'pypi', 'package_git', 'huggingface', 'postman'] RUNTIME_SOURCES = ['github', 'github_archive', 'github_archive_files', 'github_gists', 'gitlab', 'github_actions', 'gitlab_ci', 'huggingface', 'dockerhub', 'npm', 'package_git', 'postman', 'keychecks', 'pypi'] VALIDATION_SERVICES = ['anthropic', 'aws', 'azure', 'deepseek', 'dockerhub', 'gcp', 'gemini', 'github', 'gitlab', 'groq', 'huggingface', 'kimi', 'openai', 'openrouter', 'provider_resolver', 'qwen', 'replicate', 'xai', 'zai'] VALIDATION_STATUS_GROUPS = ['alive', 'dead', 'restricted', 'no_balance', 'no_context', 'limited', 'network', 'unknown'] VALIDATION_STATUSES = [ 'ALIVE', 'VALID', 'VALID_2FA', 'VALID_RATE_LIMITED', 'VERTEX', 'BEDROCK', 'FOUNDRY', 'ADMIN', 'CANARY', 'NO_BALANCE', 'NO_QUOTA', 'LIMITED_OR_NO_BALANCE', 'LIMITED_OR_QUOTA', 'LIMITED', 'RATE_LIMITED', 'NO_CONTEXT', 'NO_TARGET', 'NO_TARGET_MODELS', 'NO_GENERATION_MODEL', 'RESTRICTED', 'API_DISABLED', 'ACCESS_DENIED', 'QUARANTINED', 'DISABLED', 'DEAD', 'INVALID', 'EXPIRED', 'LEAKED_REVOKED', 'INVALID_OR_REVOKED', 'NETWORK', 'NETWORK_ERROR', 'UNKNOWN', ] VALIDATION_ACCESS_TIERS = ['usable_llm', 'alive_unproven_llm', 'no_quota', 'quota_limited', 'missing_context', 'dead', 'restricted', 'network', 'unknown'] VALIDATION_PRESETS = [ 'Usable LLM keys', 'Ever usable LLM keys', 'Ever alive / LLM candidates', 'Quota / no balance', 'Alive but not proven LLM', 'Unattributed alive', 'All latest validation', 'Latest rows', ] UNATTRIBUTED = '(unattributed)' ARCHIVE_INTERESTING_DETECTORS = { 'googleai', 'googleaistudio', 'openai', 'anthropic', 'deepseek', 'openrouter', 'groq', 'replicate', 'xai', 'huggingface', 'qwendashscope', 'kimimoonshot', 'zaiglm', 'github', 'githuboauth2', 'gitlab', 'aws', 'gcp', 'gcpapplicationdefaultcredentials', 'azure', 'azureopenai', 'azurecontainerregistry', 'dockerhub', 'npmtoken', 'sentrytoken', 'weightsandbiases', 'twilio', 'scalewaykey', 'fastlypersonaltoken', 'sonarcloud', 'honeycomb', 'rapidapi', } DASHBOARD_DB_DIALECT = 'sqlite' DASHBOARD_QUERY_TIMEOUT_SEC = 10 MAX_LOG_TAIL_BYTES = 1024 * 1024 LOOKUP_MAX_INPUT_CHARS = 65536 LOOKUP_METADATA_LIMIT = 100000 REPORTING_PRESETS = { '1 hour': timedelta(hours=1), '24 hours': timedelta(hours=24), '7 days': timedelta(days=7), '30 days': timedelta(days=30), } CORE_RUNTIME_SOURCES = { 'result-ingester', 'jsonl-projector', 'janitor', 'worker-api', 'github', 'gitlab', 'huggingface', 'dockerhub', 'package_git', 'keychecks', } FORBIDDEN_DASHBOARD_QUERY_COLUMNS = ( 'raw_secret', 'raw_result_json', 'raw_finding_json', 'config_json', 'evidence_json', ) def parse_args(): parser = argparse.ArgumentParser(add_help=False) parser.add_argument('--config') parser.add_argument('--db') parser.add_argument('--db-url') parser.add_argument('--immutable-db', action='store_true') parser.add_argument('--results-dir', default=DEFAULT_RESULTS_DIR) parser.add_argument('--queue-dir', default=DEFAULT_QUEUE_DIR) parser.add_argument('--state-file', default=DEFAULT_STATE_FILE) parser.add_argument('--log-dir', default=DEFAULT_LOG_DIR) parser.add_argument('--work-dir', default=DEFAULT_PATHS['work_dir']) parser.add_argument('--keycheck-dir', default=DEFAULT_PATHS['keycheck_dir']) parser.add_argument('--scan-limiter-db', default=os.path.join(DEFAULT_PATHS['state_dir'], 'scan_limiter.db')) parser.add_argument('--max-active-scans', type=int, default=0) args, _ = parser.parse_known_args(sys.argv[1:]) if not args.config: cwd_config = os.path.join(os.getcwd(), 'config.yaml') if os.path.exists(cwd_config): args.config = cwd_config if args.config: try: import yaml with open(args.config, 'r', encoding='utf-8') as f: config = apply_path_config(yaml.safe_load(f) or {}, args.config) global_config = config.get('global') or {} args.results_dir = global_config.get('results_dir', args.results_dir) args.queue_dir = global_config.get('queue_dir', args.queue_dir) args.state_file = global_config.get('state_file', args.state_file) args.log_dir = global_config.get('log_dir', args.log_dir) args.work_dir = global_config.get('work_dir', args.work_dir) args.keycheck_dir = global_config.get('keycheck_dir', args.keycheck_dir) args.scan_limiter_db = global_config.get('scan_limiter_db', args.scan_limiter_db) base_scans = int(global_config.get('max_active_scans', args.max_active_scans) or 0) bonus_scans = max(0, min(1, int(global_config.get('opportunistic_scan_slots', 0) or 0))) args.max_active_scans = base_scans + bonus_scans if not args.db_url: args.db_url = global_config.get('dashboard_db_url') or global_config.get('database_url') or database_url_from_env() if not args.db: args.db = resolve_optional_path(global_config.get('dashboard_db_path') or global_config.get('database_path'), global_config) elif args.db: args.db = resolve_optional_path(args.db, global_config) args.immutable_db = bool(global_config.get('dashboard_immutable_db', args.immutable_db)) except Exception as e: if os.getenv(CHILD_KIND_ENV) == 'dashboard': raise SystemExit(f'Managed dashboard config failed closed: {type(e).__name__}') from e print(f'Unable to load dashboard config {args.config}: {e}') return args def resolve_db_path(args): if args.db: return args.db return get_database_path(args.results_dir) def resolve_db_url(args): return args.db_url or database_url_from_env() def connect_db(path, db_url=None, immutable=False): global DASHBOARD_DB_DIALECT if is_postgres_url(db_url): conn = connect_postgres( db_url, connect_timeout_sec=3, statement_timeout_ms=10000, lock_timeout_ms=2000, idle_in_transaction_timeout_ms=10000, tcp_user_timeout_ms=5000, ) try: conn.execute("SET application_name = 'truf-dashboard'") conn.execute('SET default_transaction_read_only = on') conn.commit() except BaseException: conn.close() raise DASHBOARD_DB_DIALECT = 'postgres' return conn if not path or not os.path.exists(path): return None conn = connect_sqlite(path, timeout_sec=5, read_only=True, immutable=immutable, check_same_thread=False) conn.execute('PRAGMA query_only=ON') conn.execute('PRAGMA busy_timeout=5000') DASHBOARD_DB_DIALECT = 'sqlite' return conn def query_df(conn, sql, params=None): if conn is None: return pd.DataFrame() if any(re.search(rf'\b{re.escape(column)}\b', str(sql), re.IGNORECASE) for column in FORBIDDEN_DASHBOARD_QUERY_COLUMNS): st.error('Dashboard query refused because it requested a sensitive payload column.') return pd.DataFrame() raw_sqlite = getattr(conn, '_conn', None) if getattr(conn, 'is_sqlite', False) is True else None if raw_sqlite is not None: deadline = time.monotonic() + DASHBOARD_QUERY_TIMEOUT_SEC raw_sqlite.set_progress_handler(lambda: 1 if time.monotonic() >= deadline else 0, 10000) try: rows = conn.execute(sql, params or []).fetchall() if getattr(conn, 'is_postgres', False): conn.commit() return pd.DataFrame([dict(row) for row in rows]) except Exception as e: if getattr(conn, 'is_postgres', False): try: conn.rollback() except Exception: pass st.error(f'Database query failed ({type(e).__name__}). Retry after database recovery.') return pd.DataFrame() finally: if raw_sqlite is not None: raw_sqlite.set_progress_handler(None, 0) def table_columns(conn, table): if conn is None: return set() raw_sqlite = getattr(conn, '_conn', None) if getattr(conn, 'is_sqlite', False) is True else None if raw_sqlite is not None: deadline = time.monotonic() + DASHBOARD_QUERY_TIMEOUT_SEC raw_sqlite.set_progress_handler(lambda: 1 if time.monotonic() >= deadline else 0, 10000) try: columns = conn.table_columns(table) if getattr(conn, 'is_postgres', False): conn.commit() return columns except Exception: if getattr(conn, 'is_postgres', False): try: conn.rollback() except Exception: pass return set() finally: if raw_sqlite is not None: raw_sqlite.set_progress_handler(None, 0) def table_exists(conn, table): if conn is None: return False raw_sqlite = getattr(conn, '_conn', None) if getattr(conn, 'is_sqlite', False) is True else None if raw_sqlite is not None: deadline = time.monotonic() + DASHBOARD_QUERY_TIMEOUT_SEC raw_sqlite.set_progress_handler(lambda: 1 if time.monotonic() >= deadline else 0, 10000) try: exists = conn.table_exists(table) if getattr(conn, 'is_postgres', False): conn.commit() return exists except Exception: if getattr(conn, 'is_postgres', False): try: conn.rollback() except Exception: pass return False finally: if raw_sqlite is not None: raw_sqlite.set_progress_handler(None, 0) def json_extract_sql(column, path): if DASHBOARD_DB_DIALECT == 'postgres': parts = str(path or '').lstrip('$.').split('.') pg_path = ','.join(part for part in parts if part) return f"(NULLIF({column}, '')::jsonb #>> '{{{pg_path}}}')" return f"json_extract({column}, '{path}')" def today_sql(): return "CURRENT_DATE::text" if DASHBOARD_DB_DIALECT == 'postgres' else "date('now')" def finding_classification_sql(alias='f'): prefix = f'{alias}.' if alias else '' detector = f"LOWER(COALESCE({prefix}detector_name, ''))" credential = f"LOWER(COALESCE({prefix}credential_kind, ''))" return f''' CASE WHEN {detector} IN ('googleai', 'googleaistudio', 'openai', 'anthropic', 'deepseek', 'openrouter', 'groq', 'replicate', 'xai', 'huggingface', 'qwendashscope', 'qwen_dashscope', 'qwen', 'dashscope', 'kimimoonshot', 'moonshotai', 'moonshot', 'kimi', 'zaiglm', 'github', 'githuboauth2', 'gitlab', 'aws', 'gcp', 'gcpapplicationdefaultcredentials', 'azure', 'azureopenai', 'azurecontainerregistry') THEN 1 WHEN {credential} IN ('google_ai_api_key', 'google_ai_studio_api_key', 'qwen_dashscope_api_key', 'kimi_moonshot_api_key', 'glm_api_key', 'dockerhub_pat') THEN 1 ELSE 0 END ''' def finding_service_sql(alias='f'): prefix = f'{alias}.' if alias else '' detector = f"LOWER(COALESCE({prefix}detector_name, ''))" credential = f"LOWER(COALESCE({prefix}credential_kind, ''))" return f''' CASE WHEN {detector} IN ('googleai', 'googleaistudio') THEN 'gemini' WHEN {credential} IN ('google_ai_api_key', 'google_ai_studio_api_key') THEN 'gemini' WHEN {detector} = 'openai' THEN 'openai' WHEN {detector} = 'anthropic' THEN 'anthropic' WHEN {detector} = 'deepseek' THEN 'deepseek' WHEN {detector} = 'openrouter' THEN 'openrouter' WHEN {detector} = 'groq' THEN 'groq' WHEN {detector} = 'replicate' THEN 'replicate' WHEN {detector} = 'xai' THEN 'xai' WHEN {detector} = 'huggingface' THEN 'huggingface' WHEN {detector} IN ('qwendashscope', 'qwen_dashscope', 'qwen', 'dashscope') THEN 'qwen' WHEN {credential} = 'qwen_dashscope_api_key' THEN 'qwen' WHEN {detector} IN ('kimimoonshot', 'moonshotai', 'moonshot', 'kimi') THEN 'kimi' WHEN {credential} = 'kimi_moonshot_api_key' THEN 'kimi' WHEN {detector} = 'zaiglm' THEN 'zai' WHEN {credential} = 'glm_api_key' THEN 'zai' WHEN {detector} IN ('github', 'githuboauth2') THEN 'github' WHEN {detector} = 'gitlab' THEN 'gitlab' WHEN {detector} = 'aws' THEN 'aws' WHEN {detector} IN ('gcp', 'gcpapplicationdefaultcredentials') THEN 'gcp' WHEN {detector} IN ('azure', 'azureopenai', 'azurecontainerregistry') THEN 'azure' WHEN {credential} = 'dockerhub_pat' THEN 'dockerhub' ELSE 'noise' END ''' def finding_noise_reason_sql(alias='f'): prefix = f'{alias}.' if alias else '' detector = f"LOWER(COALESCE({prefix}detector_name, ''))" secret_hash = f"COALESCE({prefix}secret_hash, '')" return f''' CASE WHEN {finding_classification_sql(alias)} = 1 THEN 'keycheckable' WHEN {detector} = 'dockerhub' THEN 'dockerhub_non_pat_or_missing_username' WHEN {detector} IN ('uri', 'jdbc', 'postgres', 'mongodb', 'sqlserver') THEN 'connection_string_or_url' WHEN {detector} IN ('box', 'circle', 'flatio', 'roaring', 'linkpreview') THEN 'generic_detector_noise' WHEN {secret_hash} = '' THEN 'missing_secret_identity' ELSE 'no_checker_or_generic_secret' END ''' def keycheckable_findings_view_sql(): return f''' SELECT f.id, f.source, f.query, f.detector_name, f.secret_hash, {finding_classification_sql('f')} AS is_keycheckable, {finding_service_sql('f')} AS validation_service, {finding_noise_reason_sql('f')} AS noise_reason FROM findings f ''' def keycheckable_backlog_sql(where=''): where_clause = f'WHERE {where}' if where else '' latest_keychecks = latest_keycheck_view_sql() return f''' WITH classified AS ({keycheckable_findings_view_sql()}), checked_hashes AS ( SELECT service, COALESCE(NULLIF(secret_hash, ''), NULLIF(key_hash, ''), NULLIF(key_masked, '')) AS checked_hash, MAX(CASE WHEN status_group = 'alive' THEN 1 ELSE 0 END) AS has_alive FROM ({latest_keychecks}) WHERE COALESCE(NULLIF(secret_hash, ''), NULLIF(key_hash, ''), NULLIF(key_masked, '')) IS NOT NULL GROUP BY service, checked_hash ) SELECT source, query, validation_service, detector_name, COUNT(*) AS raw_findings, COUNT(DISTINCT NULLIF(secret_hash, '')) AS unique_secrets, COUNT(DISTINCT CASE WHEN is_keycheckable = 1 THEN NULLIF(secret_hash, '') END) AS keycheckable_unique, COUNT(DISTINCT CASE WHEN is_keycheckable = 0 THEN NULLIF(secret_hash, '') END) AS noise_unique, COUNT(DISTINCT CASE WHEN is_keycheckable = 1 AND checked_hashes.checked_hash IS NOT NULL THEN NULLIF(secret_hash, '') END) AS checked_unique, COUNT(DISTINCT CASE WHEN is_keycheckable = 1 AND checked_hashes.has_alive = 1 THEN NULLIF(secret_hash, '') END) AS alive_unique, COUNT(DISTINCT CASE WHEN is_keycheckable = 1 AND checked_hashes.checked_hash IS NULL THEN NULLIF(secret_hash, '') END) AS pending_unique FROM classified LEFT JOIN checked_hashes ON checked_hashes.service = classified.validation_service AND checked_hashes.checked_hash = classified.secret_hash {where_clause} GROUP BY source, query, validation_service, detector_name ''' def scalar(conn, sql, params=None, default=0): df = query_df(conn, sql, params) if df.empty: return default return df.iloc[0, 0] def current_queue_counts(queue_dir): rows = [] for source in SOURCES: platform = 'docker' if source == 'dockerhub' else source todo_file = os.path.join(queue_dir, f'todo_{platform}.txt') checked_file = os.path.join(queue_dir, f'checked_{platform}.txt') counts = queue_counts(todo_file, checked_file) if counts['todo_count'] or counts['checked_count'] or os.path.exists(todo_file) or os.path.exists(checked_file): rows.append({ 'source': source, 'todo': counts['todo_count'], 'checked': counts['checked_count'], 'todo_file': todo_file, 'checked_file': checked_file, }) return pd.DataFrame(rows) def queue_backlog(queue_dir): df = current_queue_counts(queue_dir) if df.empty: return 0 return int(df['todo'].fillna(0).sum()) def disk_free_gb(path): target = path if path and os.path.exists(path) else os.path.abspath(os.path.splitdrive(path or os.getcwd())[0] + os.sep) try: return round(shutil.disk_usage(target).free / (1024 ** 3), 2) except OSError: return 0 def scan_slots_df(scan_limiter_db): if not scan_limiter_db or not os.path.exists(scan_limiter_db): return pd.DataFrame() try: uri = 'file:' + scan_limiter_db.replace('\\', '/') + '?mode=ro' conn = sqlite3.connect(uri, uri=True) conn.row_factory = sqlite3.Row columns = {row[1] for row in conn.execute('PRAGMA table_info(scan_slots)').fetchall()} slot_kind = "COALESCE(slot_kind, 'base') AS slot_kind" if 'slot_kind' in columns else "'base' AS slot_kind" rows = pd.read_sql_query(f''' SELECT owner_source, owner_pid, owner_thread, {slot_kind}, command, acquired_at, updated_at FROM scan_slots ORDER BY acquired_at ''', conn) conn.close() if not rows.empty: now = datetime.now().timestamp() rows['age_sec'] = (now - rows['acquired_at']).round(0).astype(int) return rows except Exception: return pd.DataFrame() def load_runner_state(state_file): path = state_file if not os.path.exists(path): return {}, path try: with open(path, 'r', encoding='utf-8') as f: return json.load(f), path except (OSError, json.JSONDecodeError): return {}, path def parse_datetime(value): if not value: return None text = str(value).strip() for candidate in (text, text.replace('Z', '+00:00')): try: return datetime.fromisoformat(candidate) except ValueError: continue return None def human_age(value): dt = parse_datetime(value) if not dt: return '' now = datetime.now(dt.tzinfo) if dt.tzinfo else datetime.now() seconds = max(0, int((now - dt).total_seconds())) if seconds < 60: return f'{seconds}s ago' minutes = seconds // 60 if minutes < 60: return f'{minutes}m ago' hours = minutes // 60 if hours < 48: return f'{hours}h {minutes % 60}m ago' days = hours // 24 return f'{days}d {hours % 24}h ago' def read_tail(path, lines=80): if not path or not os.path.exists(path): return [] max_lines = max(1, int(lines or 80)) try: with open(path, 'rb') as f: f.seek(0, os.SEEK_END) size = f.tell() position = max(0, size - MAX_LOG_TAIL_BYTES) f.seek(position) data = f.read(MAX_LOG_TAIL_BYTES) if position and data: newline = data.find(b'\n') data = data[newline + 1:] if newline >= 0 else b'' text = data.decode('utf-8', errors='replace') return text.splitlines(keepends=True)[-max_lines:] except OSError: return [] def parse_supervisor_status(log_dir): path = os.path.join(log_dir, 'supervisor.status.txt') rows = [] if not os.path.exists(path): return pd.DataFrame(rows), path headers = None for line in read_tail(path, 80): if '|' not in line or line.lstrip().startswith('-'): continue parts = [part.strip() for part in line.split('|')] lowered = [part.lower() for part in parts] if lowered and lowered[0] == 'source' and 'status' in lowered: headers = lowered continue if headers and len(parts) >= len(headers): item = dict(zip(headers, parts)) rows.append({ 'source': item.get('source', ''), 'runtime_status': item.get('status', ''), 'desired': item.get('desired', ''), 'pid': item.get('pid', ''), 'mode': item.get('mode', ''), 'up': item.get('up', ''), 'exit': item.get('exit', ''), 'next': item.get('next', ''), 'restarts': item.get('rs', item.get('restarts', '')), 'auth': item.get('auth', ''), }) continue if len(parts) < 9 or parts[0] in ('', 'Type `help` for commands. Use `command ` for full log/state paths.'): continue source = parts[0] if source == 'source': continue has_auth_column = len(parts) >= 10 rows.append({ 'source': source, 'runtime_status': parts[1], 'desired': '', 'pid': parts[2], 'mode': parts[3], 'up': parts[4], 'exit': parts[5], 'next': parts[6], 'restarts': parts[7], 'auth': parts[8] if has_auth_column else '', }) return pd.DataFrame(rows), path def load_per_source_states(log_dir): state_dir = os.path.normpath(os.path.join(log_dir, '..', 'state')) rows = [] for source in RUNTIME_SOURCES: path = os.path.join(state_dir, f'runner_state_{source}.json') if not os.path.exists(path): continue try: with open(path, 'r', encoding='utf-8') as f: state = json.load(f) except (OSError, json.JSONDecodeError): continue item = (state.get('sources') or {}).get(source) or {} rows.append({ 'source': source, 'last_query': item.get('last_query'), 'state_status': item.get('last_status'), 'last_started_at': item.get('last_started_at'), 'last_started_age': human_age(item.get('last_started_at')), 'last_completed_at': item.get('last_completed_at'), 'last_completed_age': human_age(item.get('last_completed_at')), 'cycles': item.get('cycles'), 'last_scanned': item.get('last_scanned'), 'last_auth': item.get('last_auth'), }) return pd.DataFrame(rows) def latest_cycle_df(conn): return query_df(conn, ''' SELECT source, status AS db_status, query AS db_query, started_at AS db_started_at, ended_at AS db_ended_at, scanned_count AS db_scanned, findings_count AS db_findings, error_count AS db_errors, skipped_count AS db_skipped, duration_sec AS db_duration_sec FROM source_cycles WHERE id IN (SELECT MAX(id) FROM source_cycles GROUP BY source) ''') def runtime_source_health(conn, log_dir): runtime_df, status_path = parse_supervisor_status(log_dir) states_df = load_per_source_states(log_dir) latest_df = latest_cycle_df(conn) sources = sorted(set(RUNTIME_SOURCES) | set(runtime_df['source'].tolist() if not runtime_df.empty else []) | set(states_df['source'].tolist() if not states_df.empty else []) | set(latest_df['source'].tolist() if not latest_df.empty else [])) rows = pd.DataFrame({'source': sources}) for df in (runtime_df, states_df, latest_df): if not df.empty: rows = rows.merge(df, on='source', how='left') log_rows = [] for source in sources: log_path = os.path.join(log_dir, f'{source}.log') log_rows.append({ 'source': source, 'log_path': log_path if os.path.exists(log_path) else '', 'log_updated': datetime.fromtimestamp(os.path.getmtime(log_path)).isoformat(timespec='seconds') if os.path.exists(log_path) else '', 'log_age': human_age(datetime.fromtimestamp(os.path.getmtime(log_path)).isoformat(timespec='seconds')) if os.path.exists(log_path) else '', 'log_bytes': os.path.getsize(log_path) if os.path.exists(log_path) else 0, }) rows = rows.merge(pd.DataFrame(log_rows), on='source', how='left') def classify(row): runtime = str(row.get('runtime_status') or '').lower() if runtime in ('running', 'waiting', 'blocked', 'paused', 'backoff', 'done', 'failed', 'stopped', 'disabled'): return runtime if row.get('log_path'): return 'has logs' return 'no data' rows['health'] = rows.apply(classify, axis=1) preferred = [ 'source', 'health', 'runtime_status', 'desired', 'pid', 'mode', 'up', 'exit', 'next', 'restarts', 'auth', 'state_status', 'last_query', 'last_started_age', 'last_completed_age', 'last_scanned', 'db_status', 'db_query', 'db_scanned', 'db_findings', 'db_errors', 'log_age', 'log_bytes', ] existing = [column for column in preferred if column in rows.columns] return rows[existing], status_path def show_metrics(metrics): cols = st.columns(len(metrics)) for col, (label, value) in zip(cols, metrics): col.metric(label, value) def format_pct(value): try: return f'{float(value) * 100:.2f}%' except (TypeError, ValueError): return '0.00%' def display_df(df, height=None): if df.empty: st.info('No data yet') return safe_secret_metadata = { 'redacted_secret', 'secret_hash', 'detector_secret_hash', 'unique_secrets', 'unique_secrets_count', 'secrets_found', 'credential_kind', 'credential_confidence', 'required_context_missing', } blocked = [] for column in df.columns: name = str(column).lower() if name in safe_secret_metadata: continue if ( name.startswith('raw') or name in {'config_json', 'evidence_json', 'credential', 'credential_value'} or 'password' in name or name == 'token' or name.endswith('_token') ): blocked.append(column) df = df.drop(columns=blocked, errors='ignore') endpoint_columns = [ column for column in df.columns if str(column).lower() == 'endpoint' or str(column).lower().endswith('_endpoint') ] if endpoint_columns: df = df.copy() for column in endpoint_columns: df[column] = df[column].map( lambda value: value if pd.isna(value) else sanitize_endpoint(value) ) if 'resource' in df.columns: adc_rows = pd.Series(False, index=df.index) for detector_column in ('detector_name', 'detector'): if detector_column in df.columns: adc_rows |= df[detector_column].fillna('').astype(str).str.lower().eq( 'gcpapplicationdefaultcredentials' ) if 'credential_kind' in df.columns: adc_rows |= df['credential_kind'].fillna('').astype(str).str.lower().eq( 'application_default_credentials' ) if adc_rows.any(): df = df.copy() df.loc[adc_rows, 'resource'] = '' if 'redacted_secret' in df.columns: df = df.copy() df['redacted_secret'] = df['redacted_secret'].map( lambda value: '***REDACTED***' if pd.notna(value) and str(value) else '' ) if height is None: st.dataframe(df, width='stretch') else: st.dataframe(df, width='stretch', height=height) def latest_keycheck_view_sql(): return ''' SELECT kr.*, COALESCE(NULLIF(kr.key_hash, ''), NULLIF(kr.secret_hash, ''), NULLIF(kr.key_masked, '')) AS key_identity, 1 AS latest_rank FROM keycheck_current_state state JOIN keycheck_results kr ON kr.id = state.last_result_id ''' def validation_access_tier_sql(alias='kr'): prefix = f'{alias}.' if alias else '' service = f"LOWER(COALESCE({prefix}service, ''))" status = f"UPPER(COALESCE({prefix}status, ''))" group = f"LOWER(COALESCE({prefix}status_group, ''))" metadata = f"COALESCE({prefix}metadata_json, '{{}}')" llm_probe_status = json_extract_sql(metadata, '$.llm_probe_status') probe_status = json_extract_sql(metadata, '$.probe.status') deployment_count = json_extract_sql(metadata, '$.deployment_count') route_probe = json_extract_sql(metadata, '$.route_probe') foundry_route_probe = json_extract_sql(metadata, '$.foundry_route_probe') return f''' CASE WHEN {group} = 'no_balance' THEN 'no_quota' WHEN {status} IN ('LIMITED', 'RATE_LIMITED', 'VALID_RATE_LIMITED') THEN 'quota_limited' WHEN {service} = 'openai' AND {status} = 'ALIVE' THEN 'usable_llm' WHEN {service} IN ('anthropic', 'deepseek', 'kimi', 'openrouter') AND {status} = 'VALID' THEN 'usable_llm' WHEN {service} IN ('groq', 'qwen', 'xai') AND {status} = 'VALID' AND {llm_probe_status} = 'GENERATION_OK' THEN 'usable_llm' WHEN {service} = 'gemini' AND {status} = 'VALID' AND {probe_status} = 'GENERATION_OK' THEN 'usable_llm' WHEN {service} = 'gcp' AND {status} = 'VERTEX' THEN 'usable_llm' WHEN {service} = 'aws' AND {status} = 'BEDROCK' THEN 'usable_llm' WHEN {service} = 'azure' AND {status} = 'VALID' AND COALESCE(CAST({deployment_count} AS INTEGER), 0) > 0 AND {route_probe} = 'accepted_auth_route' THEN 'usable_llm' WHEN {service} = 'azure' AND {status} = 'FOUNDRY' AND {foundry_route_probe} = 'accepted' THEN 'usable_llm' WHEN {group} = 'alive' THEN 'alive_unproven_llm' WHEN {group} = 'limited' THEN 'quota_limited' WHEN {group} = 'no_context' THEN 'missing_context' ELSE {group} END ''' def as_utc(value): if not isinstance(value, datetime): raise ValueError('Reporting timestamps must be datetime values.') if value.tzinfo is None: return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) def reporting_window(preset, now=None, custom_start=None, custom_end=None): end = as_utc(now or datetime.now(timezone.utc)) if preset in REPORTING_PRESETS: start = end - REPORTING_PRESETS[preset] elif preset == 'Custom': if custom_start is None or custom_end is None: raise ValueError('Choose both custom UTC timestamps.') start = as_utc(custom_start) end = as_utc(custom_end) else: raise ValueError('Unknown reporting window.') if end <= start: raise ValueError('The end of the reporting window must be later than the start.') return start, end def utc_parameter(value): return as_utc(value).isoformat(timespec='seconds') def _secret_from_payload(value, depth=0): if depth > 4: return None if isinstance(value, dict): lowered = {str(key).lower(): item for key, item in value.items()} for name in ('rawv2', 'raw_v2', 'raw', 'secret', 'api_key', 'apikey', 'access_token'): candidate = lowered.get(name) if isinstance(candidate, str) and candidate.strip(): return candidate.strip() for item in value.values(): candidate = _secret_from_payload(item, depth + 1) if candidate: return candidate elif isinstance(value, list): for item in value[:100]: candidate = _secret_from_payload(item, depth + 1) if candidate: return candidate return None def _credential_like(text): lowered = text.lower() known_prefixes = ( 'sk-', 'sk_', 'ghp_', 'github_pat_', 'glpat-', 'hf_', 'gsk_', 'xai-', 'sk-or-', 'r8_', 'npm_', 'akia', 'asia', 'aiza', 'eyj', ) if lowered.startswith(known_prefixes): return True if len(text) < 20 or any(character.isspace() for character in text): return False if text.startswith(('http://', 'https://', 'git@')) or any(character in text for character in ('/', '\\', '@')): return False if re.fullmatch(r'[0-9a-fA-F]{40}', text) or re.fullmatch( r'[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}', text): return False return bool(re.fullmatch(r'[A-Za-z0-9_.:+\-=]+', text)) def normalize_lookup(text): value = str(text or '').strip() if not value: raise ValueError('Paste a key, hash, finding ID/UID, target, path, or commit.') if len(value) > LOOKUP_MAX_INPUT_CHARS: raise ValueError('Lookup input is too large.') extracted = None if value[:1] in ('{', '['): try: extracted = _secret_from_payload(json.loads(value)) except (TypeError, ValueError, json.JSONDecodeError): extracted = None if not extracted and ('\n' in value or '\r' in value): match = re.search(r'(?im)^\s*(?:rawv2|raw|secret|api[_ -]?key)\s*[:=]\s*["\']?([^\s"\']+)', value) extracted = match.group(1) if match else None if extracted: return {'kind': 'digest', 'digest': hashlib.sha256(extracted.encode('utf-8')).hexdigest()} finding_id = re.fullmatch(r'(?i)(?:finding\s*[:#]?\s*)?(\d+)', value) if finding_id: return {'kind': 'finding_id', 'finding_id': int(finding_id.group(1))} if re.fullmatch(r'[0-9a-fA-F]{64}', value): return {'kind': 'digest', 'digest': value.lower()} if re.fullmatch(r'[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}', value): return {'kind': 'identity', 'identity': value} if _credential_like(value): return {'kind': 'digest', 'digest': hashlib.sha256(value.encode('utf-8')).hexdigest()} if '\n' in value or '\r' in value: raise ValueError('Could not identify a credential in the pasted finding.') if len(value) < 3: raise ValueError('Metadata lookup needs at least three characters.') if len(value) > 512: raise ValueError('Metadata lookup is limited to 512 characters.') return {'kind': 'metadata', 'text': value} def escaped_like(value): return '%' + str(value).replace('!', '!!').replace('%', '!%').replace('_', '!_') + '%' def period_summary_df(conn, start, end): access_tier = validation_access_tier_sql('r') return query_df(conn, f''' WITH bounds AS ( SELECT CAST(? AS timestamptz) AS start_at, CAST(? AS timestamptz) AS end_at ), period_scans AS ( SELECT COUNT(*) AS scans, COALESCE(SUM(ts.findings_count), 0) AS findings, COALESCE(SUM(ts.error_count), 0) AS errors FROM target_scans ts CROSS JOIN bounds b WHERE ts.ended_at::timestamptz >= b.start_at AND ts.ended_at::timestamptz < b.end_at ), period_checks AS ( SELECT COUNT(*) AS checks FROM keycheck_results kr CROSS JOIN bounds b WHERE kr.checked_at::timestamptz >= b.start_at AND kr.checked_at::timestamptz < b.end_at ), period_discovery AS ( SELECT COALESCE(SUM(sc.queued_new_count), 0) AS queued_new, COALESCE(SUM(sc.queued_updated_count), 0) AS queued_updated FROM source_cycles sc CROSS JOIN bounds b WHERE sc.started_at::timestamptz >= b.start_at AND sc.started_at::timestamptz < b.end_at ), ranked_alive AS ( SELECT r.id, r.credential_id, r.checked_at, ROW_NUMBER() OVER ( PARTITION BY r.credential_id ORDER BY r.checked_at::timestamptz, r.id ) AS alive_rank FROM keycheck_results r WHERE r.credential_id IS NOT NULL AND r.status_group = 'alive' ), ranked_usable AS ( SELECT r.id, r.credential_id, r.checked_at, ROW_NUMBER() OVER ( PARTITION BY r.credential_id ORDER BY r.checked_at::timestamptz, r.id ) AS usable_rank FROM keycheck_results r WHERE r.credential_id IS NOT NULL AND ({access_tier}) = 'usable_llm' ), period_new_alive AS ( SELECT COUNT(*) AS credentials FROM ranked_alive r CROSS JOIN bounds b WHERE r.alive_rank = 1 AND r.checked_at::timestamptz >= b.start_at AND r.checked_at::timestamptz < b.end_at ), period_new_usable AS ( SELECT COUNT(*) AS credentials FROM ranked_usable r CROSS JOIN bounds b WHERE r.usable_rank = 1 AND r.checked_at::timestamptz >= b.start_at AND r.checked_at::timestamptz < b.end_at ) SELECT ps.scans, ps.findings, ps.errors, pc.checks, pd.queued_new, pd.queued_updated, pna.credentials AS new_alive, pnu.credentials AS new_usable, (SELECT COUNT(*) FROM keycheck_current_state WHERE status_group = 'alive') AS alive_now, (SELECT COUNT(*) FROM keycheck_current_state s JOIN keycheck_results r ON r.id = s.last_result_id WHERE s.status_group = 'alive' AND ({access_tier}) = 'usable_llm') AS usable_llm_now, (SELECT COUNT(*) FROM target_queue WHERE status IN ('pending', 'deferred')) AS target_backlog, (SELECT COUNT(*) FROM keycheck_candidates WHERE state IN ('pending', 'deferred', 'leased')) AS keycheck_backlog FROM period_scans ps CROSS JOIN period_checks pc CROSS JOIN period_discovery pd CROSS JOIN period_new_alive pna CROSS JOIN period_new_usable pnu ''', [utc_parameter(start), utc_parameter(end)]) def source_activity_df(conn, start, end): return query_df(conn, ''' WITH bounds AS ( SELECT CAST(? AS timestamptz) AS start_at, CAST(? AS timestamptz) AS end_at ), scan_activity AS ( SELECT ts.source, COUNT(*) AS scans, COALESCE(SUM(ts.findings_count), 0) AS findings, COALESCE(SUM(ts.error_count), 0) AS errors, MAX(ts.ended_at) AS latest_scan FROM target_scans ts CROSS JOIN bounds b WHERE ts.ended_at::timestamptz >= b.start_at AND ts.ended_at::timestamptz < b.end_at GROUP BY ts.source ), cycle_activity AS ( SELECT sc.source, COALESCE(SUM(sc.queued_new_count), 0) AS queued_new, COALESCE(SUM(sc.queued_updated_count), 0) AS queued_updated, MAX(sc.started_at) AS latest_cycle FROM source_cycles sc CROSS JOIN bounds b WHERE sc.started_at::timestamptz >= b.start_at AND sc.started_at::timestamptz < b.end_at GROUP BY sc.source ), sources AS ( SELECT source FROM scan_activity UNION SELECT source FROM cycle_activity ) SELECT s.source, COALESCE(sa.scans, 0) AS scans, COALESCE(sa.findings, 0) AS findings, COALESCE(sa.errors, 0) AS errors, COALESCE(ca.queued_new, 0) AS queued_new, COALESCE(ca.queued_updated, 0) AS queued_updated, GREATEST(sa.latest_scan, ca.latest_cycle) AS latest FROM sources s LEFT JOIN scan_activity sa ON sa.source = s.source LEFT JOIN cycle_activity ca ON ca.source = s.source ORDER BY findings DESC, scans DESC, s.source LIMIT 50 ''', [utc_parameter(start), utc_parameter(end)]) def new_alive_breakdown_df(conn, start, end): access_tier = validation_access_tier_sql('kr') return query_df(conn, f''' WITH bounds AS ( SELECT CAST(? AS timestamptz) AS start_at, CAST(? AS timestamptz) AS end_at ), ranked_alive AS ( SELECT kr.credential_id AS id, kr.service, kr.status, kr.checked_at, ROW_NUMBER() OVER ( PARTITION BY kr.credential_id ORDER BY kr.checked_at::timestamptz, kr.id ) AS alive_rank, {access_tier} AS access_tier FROM keycheck_results kr WHERE kr.credential_id IS NOT NULL AND kr.status_group = 'alive' ), recent AS ( SELECT ra.id, ra.service, ra.status, ra.checked_at, ra.access_tier FROM ranked_alive ra CROSS JOIN bounds b WHERE ra.alive_rank = 1 AND ra.checked_at::timestamptz >= b.start_at AND ra.checked_at::timestamptz < b.end_at ) SELECT r.service, r.status, r.access_tier, COALESCE(origin.source, '(unknown)') AS source, COUNT(*) AS credentials, MAX(r.checked_at) AS latest FROM recent r LEFT JOIN LATERAL ( SELECT kc.source FROM keycheck_candidates kc WHERE kc.credential_id = r.id ORDER BY kc.id LIMIT 1 ) origin ON TRUE GROUP BY r.service, r.status, r.access_tier, origin.source ORDER BY credentials DESC, r.service, r.status, r.access_tier, source LIMIT 100 ''', [utc_parameter(start), utc_parameter(end)]) def pipeline_snapshot_df(conn): return query_df(conn, ''' SELECT bundle_items, projection_items, keycheck_items, quarantine_items, bundle_bytes, projection_bytes, keycheck_bytes, quarantine_bytes, updated_at FROM pipeline_capacity WHERE id = 1 ''') def lookup_findings_df(conn, lookup, limit=200): kind = lookup.get('kind') params = [] if kind == 'finding_id': where = 'f.id = ?' params.append(int(lookup['finding_id'])) elif kind == 'digest': where = '''( f.secret_hash = ? OR f.detector_secret_hash = ? OR f.finding_uid = ? OR f.finding_fingerprint = ? )''' params.extend([lookup['digest']] * 4) elif kind == 'identity': where = '(f.finding_uid = ? OR f.finding_fingerprint = ?)' params.extend([lookup['identity']] * 2) elif kind == 'metadata': pattern = escaped_like(lookup['text']) where = ''' f.id >= (SELECT GREATEST(COALESCE(MAX(id), 0) - ?, 0) FROM findings) AND ( f.source ILIKE ? ESCAPE '!' OR f.query ILIKE ? ESCAPE '!' OR f.target ILIKE ? ESCAPE '!' OR f.detector_name ILIKE ? ESCAPE '!' OR f.redacted_secret ILIKE ? ESCAPE '!' OR f.file_path ILIKE ? ESCAPE '!' OR f.commit_hash ILIKE ? ESCAPE '!' OR f.provider ILIKE ? ESCAPE '!' OR f.credential_kind ILIKE ? ESCAPE '!' OR ts.package_name ILIKE ? ESCAPE '!' OR ts.package_version ILIKE ? ESCAPE '!' OR ts.package_filename ILIKE ? ESCAPE '!' ) ''' params.extend([LOOKUP_METADATA_LIMIT, *([pattern] * 12)]) else: return pd.DataFrame() params.append(max(1, min(int(limit or 200), 500))) return query_df(conn, f''' SELECT f.id AS finding_id, f.finding_uid, f.created_at AS found_at, f.source, f.query, f.target, f.detector_name, f.provider, f.file_path, f.line_number, f.commit_hash, f.secret_hash, ts.package_name, ts.package_version, ts.package_filename, ts.package_type FROM findings f LEFT JOIN target_scans ts ON ts.id = f.target_scan_id WHERE {where} ORDER BY f.id DESC LIMIT ? ''', params) def lookup_status_df(conn, lookup, findings=None, limit=100): findings = findings if isinstance(findings, pd.DataFrame) else pd.DataFrame() hashes = set() finding_ids = set() if not findings.empty: if 'secret_hash' in findings: hashes.update(str(value) for value in findings['secret_hash'].dropna() if str(value)) if 'finding_id' in findings: finding_ids.update(int(value) for value in findings['finding_id'].dropna()) if lookup.get('kind') == 'digest': hashes.add(lookup['digest']) if lookup.get('kind') == 'finding_id': finding_ids.add(int(lookup['finding_id'])) clauses = [] params = [] if hashes: values = sorted(hashes) placeholders = ','.join('?' for _ in values) clauses.append(f"(r.secret_hash IN ({placeholders}) OR r.key_hash IN ({placeholders}))") params.extend(values) params.extend(values) clauses.append(f'''EXISTS ( SELECT 1 FROM keycheck_candidates kc WHERE kc.credential_id = s.credential_id AND kc.secret_hash IN ({placeholders}) )''') params.extend(values) if finding_ids: values = sorted(finding_ids) placeholders = ','.join('?' for _ in values) clauses.append(f'r.finding_id IN ({placeholders})') params.extend(values) clauses.append(f'''EXISTS ( SELECT 1 FROM keycheck_candidates kc WHERE kc.credential_id = s.credential_id AND kc.finding_id IN ({placeholders}) )''') params.extend(values) if lookup.get('kind') == 'metadata': pattern = escaped_like(lookup['text']) clauses.append('''( r.key_masked ILIKE ? ESCAPE '!' OR r.source ILIKE ? ESCAPE '!' OR r.query ILIKE ? ESCAPE '!' OR r.target ILIKE ? ESCAPE '!' OR r.detector_name ILIKE ? ESCAPE '!' )''') params.extend([pattern] * 5) if not clauses: return pd.DataFrame() access_tier = validation_access_tier_sql('r') params.append(max(1, min(int(limit or 100), 200))) return query_df(conn, f''' SELECT s.credential_id, s.service, s.status, s.status_group, {access_tier} AS access_tier, s.checked_at, s.recheck_after, r.key_masked, COALESCE(r.source, origin.source) AS source, COALESCE(r.query, origin.query) AS query, COALESCE(r.target, origin.target) AS target, COALESCE(r.finding_id, origin.finding_id) AS finding_id, COALESCE(r.detector_name, origin.detector_name) AS detector_name, COALESCE(r.found_at, origin.found_at) AS found_at FROM keycheck_current_state s JOIN keycheck_results r ON r.id = s.last_result_id LEFT JOIN LATERAL ( SELECT kc.source, kc.query, kc.target, kc.finding_id, kc.detector_name, kc.found_at FROM keycheck_candidates kc WHERE kc.credential_id = s.credential_id ORDER BY (kc.finding_id IS NOT NULL) DESC, kc.id DESC LIMIT 1 ) origin ON TRUE WHERE {' OR '.join(f'({clause})' for clause in clauses)} ORDER BY s.checked_at DESC, s.credential_id LIMIT ? ''', params) def dashboard_search_is_safe(text): text = str(text or '').strip() return bool( not text or re.fullmatch(r'[0-9a-fA-F]{64}', text) or len(text) < 20 or any(character.isspace() for character in text) or '***' in text or '...' in text ) def finding_metadata_search(conn, text, limit=500): text = str(text or '').strip() if not text or not dashboard_search_is_safe(text): return pd.DataFrame() digest = text.lower() if re.fullmatch(r'[0-9a-fA-F]{64}', text) else '' like = f'%{text}%' hash_clause = 'f.secret_hash = ? OR' if digest else '' params = ([digest] if digest else []) + [like] * 9 + [int(limit)] return query_df(conn, f''' SELECT f.id AS finding_id, f.created_at, f.source, f.query, f.target, f.detector_name, f.redacted_secret, f.secret_hash, f.file_path, f.line_number, f.commit_hash, f.source_timestamp, ts.package_name, ts.package_version, ts.package_filename, ts.package_type, f.target_scan_id, f.cycle_id, f.run_id FROM ( SELECT id, created_at, source, query, target, detector_name, redacted_secret, secret_hash, file_path, line_number, commit_hash, source_timestamp, target_scan_id, cycle_id, run_id FROM findings ORDER BY id DESC LIMIT 50000 ) f LEFT JOIN target_scans ts ON ts.id = f.target_scan_id WHERE {hash_clause} f.redacted_secret LIKE ? OR f.secret_hash LIKE ? OR f.detector_name LIKE ? OR f.source LIKE ? OR f.query LIKE ? OR f.target LIKE ? OR f.file_path LIKE ? OR f.commit_hash LIKE ? OR ts.package_name LIKE ? ORDER BY f.id DESC LIMIT ? ''', params) def github_archive_yield(conn, limit=2000): rows = query_df(conn, ''' SELECT id, target, ended_at, findings_count, verified_findings_count FROM target_scans WHERE source = 'github_archive' AND status = 'found' ORDER BY id DESC LIMIT ? ''', [int(limit or 2000)]) if rows.empty: return {}, pd.DataFrame(), pd.DataFrame() scan_ids = [int(value) for value in rows['id'].dropna().tolist()] finding_rows = query_df(conn, f''' SELECT target_scan_id, detector_name FROM findings WHERE target_scan_id IN ({','.join('?' for _ in scan_ids)}) ''', scan_ids) if scan_ids else pd.DataFrame() findings_by_scan = {} if not finding_rows.empty: for scan_id, group in finding_rows.groupby('target_scan_id'): findings_by_scan[int(scan_id)] = [str(value or '') for value in group['detector_name'].tolist()] detectors = Counter() target_rows = [] total_findings = 0 total_verified = 0 interesting_rows = 0 for _, row in rows.iterrows(): findings_count = int(row.get('findings_count') or 0) verified_count = int(row.get('verified_findings_count') or 0) total_findings += findings_count total_verified += verified_count target_interesting = 0 for detector in findings_by_scan.get(int(row.get('id') or 0), []): detectors[detector] += 1 if detector.lower() in ARCHIVE_INTERESTING_DETECTORS: interesting_rows += 1 target_interesting += 1 target_rows.append({ 'target_scan_id': int(row.get('id') or 0), 'target': row.get('target'), 'ended_at': row.get('ended_at'), 'findings': findings_count, 'interesting_findings': target_interesting, 'verified_findings': verified_count, }) detector_df = pd.DataFrame([ { 'detector': detector, 'rows': count, 'kind': 'interesting' if str(detector).lower() in ARCHIVE_INTERESTING_DETECTORS else 'noise_or_generic', } for detector, count in detectors.most_common(50) ]) target_df = pd.DataFrame(target_rows).sort_values(['interesting_findings', 'findings'], ascending=False).head(100) summary = { 'found_targets': int(len(rows)), 'raw_findings': int(total_findings), 'interesting_findings': int(interesting_rows), 'noise_or_generic_findings': int(max(0, total_findings - interesting_rows)), 'verified_findings': int(total_verified), } return summary, detector_df, target_df def keycheck_summary_df(keycheck_dir): path = os.path.join(keycheck_dir, 'summary.tsv') if not os.path.exists(path): return pd.DataFrame(), path try: df = pd.read_csv(path, sep='\t') except Exception: return pd.DataFrame(), path if 'updated_at' in df.columns: df['file_updated_at'] = df['updated_at'] return df, path def recent_keycheck_db_summary(conn, limit=10000): result_source_expr = json_extract_sql('metadata_json', '$.result_source') rows = query_df(conn, ''' SELECT id, service, status_group, status, checked_at, created_at, {result_source_expr} AS result_source FROM keycheck_results ORDER BY id DESC LIMIT ? '''.format(result_source_expr=result_source_expr), [int(limit or 10000)]) if rows.empty: return pd.DataFrame() rows['result_source'] = rows['result_source'].fillna('api_check') grouped = rows.groupby('service', dropna=False).agg( db_recent_rows=('id', 'count'), db_latest_id=('id', 'max'), db_latest_checked=('checked_at', 'max'), db_latest_created=('created_at', 'max'), db_cached_rows=('result_source', lambda s: int((s == 'cached_status').sum())), db_alive_rows=('status_group', lambda s: int((s == 'alive').sum())), ).reset_index() return grouped def keycheck_db_metric_summary(keycheck_dir): rows = [] if not keycheck_dir or not os.path.isdir(keycheck_dir): return pd.DataFrame(rows) for service in sorted(os.listdir(keycheck_dir)): path = os.path.join(keycheck_dir, service, 'db_write_metrics.jsonl') if not os.path.exists(path): continue ok = failed = cached = total = 0 latest = '' for line in read_tail(path, 2000): try: item = json.loads(line) except ValueError: continue total += 1 ok += 1 if item.get('ok') else 0 failed += 0 if item.get('ok') else 1 cached += 1 if item.get('result_source') == 'cached_status' else 0 latest = max(latest, str(item.get('created_at') or '')) rows.append({ 'service': service, 'metric_rows_tail': total, 'db_write_ok_tail': ok, 'db_write_failed_tail': failed, 'cached_occurrence_tail': cached, 'db_metric_latest': latest, }) return pd.DataFrame(rows) def keycheck_pipeline_health(log_dir): path = os.path.join(log_dir, 'keychecks.log') lines = read_tail(path, 600) if not lines: return pd.DataFrame(), path latest_start = '' latest_exit = '' current_service = '' last_service = '' db_locks = 0 processed = 0 skipped = 0 for line in lines: text = line.strip() if text.startswith('=== supervisor start') and 'source=keychecks' in text: latest_start = text.replace('=== supervisor start ', '').split(' source=', 1)[0] current_service = '' processed = 0 skipped = 0 db_locks = 0 elif text.startswith('=== supervisor exit') and 'source=keychecks' in text: latest_exit = text.replace('=== supervisor exit ', '').split(' source=', 1)[0] current_service = '' elif ': ' in text and 'keycheckers' in text and '.py' in text: current_service = text.split(':', 1)[0] last_service = current_service elif 'Observability DB locked' in text or 'database is locked' in text: db_locks += 1 elif text.startswith('Done. Processed=') or text.startswith('Processed='): numbers = re.findall(r'(?:Processed|skipped)=([0-9]+)', text) if numbers: processed += int(numbers[0]) if len(numbers) > 1: skipped += int(numbers[1]) row = { 'latest_start': latest_start, 'latest_exit': latest_exit, 'current_or_last_service': current_service or last_service, 'db_lock_messages_tail': db_locks, 'processed_tail': processed, 'skipped_tail': skipped, 'log_path': path, } return pd.DataFrame([row]), path def _integer(value): try: if pd.isna(value): return 0 return int(value) except (TypeError, ValueError): return 0 def _submit_dashboard_lookup(): submitted = st.session_state.get('dashboard_lookup_input', '') try: st.session_state['dashboard_lookup_request'] = normalize_lookup(submitted) st.session_state['dashboard_lookup_error'] = '' except ValueError as exc: st.session_state.pop('dashboard_lookup_request', None) st.session_state['dashboard_lookup_error'] = str(exc) st.session_state['dashboard_lookup_input'] = '' def _clear_dashboard_lookup(): st.session_state.pop('dashboard_lookup_request', None) st.session_state.pop('dashboard_lookup_error', None) st.session_state['dashboard_lookup_input'] = '' def _dashboard_styles(): st.markdown(''' ''', unsafe_allow_html=True) def _metric_grid(metrics): cards = ''.join( '
' f'
{html.escape(str(label))}
' f'
{html.escape(str(value))}
' '
' for label, value in metrics ) st.markdown(f'
{cards}
', unsafe_allow_html=True) def page_simple_dashboard(conn, log_dir, work_dir, scan_limiter_db, max_active_scans): _dashboard_styles() title_col, refresh_col = st.columns([8, 1]) with title_col: st.title('TRUF') st.caption('Scanner and credential status / PostgreSQL read-only / UTC') with refresh_col: if st.button('Refresh', width='stretch'): st.rerun() st.subheader('Find a credential or finding') with st.form('dashboard_lookup_form', clear_on_submit=False, border=False): input_col, submit_col = st.columns([8, 1]) with input_col: st.text_input( 'Lookup', key='dashboard_lookup_input', placeholder='Paste a key, SHA-256, finding ID/UID, URL, path, or commit', label_visibility='collapsed', ) with submit_col: st.form_submit_button('Find', width='stretch', on_click=_submit_dashboard_lookup) st.caption('Credential-like input is hashed immediately, cleared, and never queried as raw text.') lookup_error = st.session_state.get('dashboard_lookup_error') lookup = st.session_state.get('dashboard_lookup_request') if lookup_error: st.warning(lookup_error) if lookup: findings = lookup_findings_df(conn, lookup) statuses = lookup_status_df(conn, lookup, findings) result_col, clear_col = st.columns([8, 1]) with result_col: st.markdown(f'**Lookup result:** {len(statuses)} current status row(s), {len(findings)} origin(s)') with clear_col: st.button('Clear', width='stretch', on_click=_clear_dashboard_lookup) if statuses.empty and findings.empty: st.info('No current status or finding origin matched this lookup.') if not statuses.empty: st.markdown('**Current status**') status_columns = [ 'service', 'status', 'status_group', 'access_tier', 'checked_at', 'source', 'query', 'target', 'detector_name', 'finding_id', 'found_at', ] display_df(statuses[[column for column in status_columns if column in statuses]], height=260) if not findings.empty: st.markdown('**Origins**') origins = findings.copy() origins['identity'] = origins['secret_hash'].map( lambda value: (str(value)[:12] + '...') if pd.notna(value) and str(value) else '' ) origin_columns = [ 'finding_id', 'identity', 'found_at', 'source', 'query', 'target', 'detector_name', 'provider', 'file_path', 'line_number', 'commit_hash', 'package_name', 'package_version', 'package_filename', 'finding_uid', ] display_df(origins[[column for column in origin_columns if column in origins]], height=360) st.divider() st.subheader('Activity') period = st.radio( 'Reporting window', [*REPORTING_PRESETS, 'Custom'], index=1, horizontal=True, label_visibility='collapsed', ) now = datetime.now(timezone.utc).replace(microsecond=0) custom_start = None custom_end = None if period == 'Custom': default_start = now - timedelta(hours=24) start_date_col, start_time_col, end_date_col, end_time_col = st.columns(4) with start_date_col: start_date = st.date_input('Start date (UTC)', value=default_start.date()) with start_time_col: start_time = st.time_input('Start time (UTC)', value=default_start.time()) with end_date_col: end_date = st.date_input('End date (UTC)', value=now.date()) with end_time_col: end_time = st.time_input('End time (UTC)', value=now.time()) custom_start = datetime.combine(start_date, start_time, tzinfo=timezone.utc) custom_end = datetime.combine(end_date, end_time, tzinfo=timezone.utc) try: start, end = reporting_window(period, now=now, custom_start=custom_start, custom_end=custom_end) except ValueError as exc: st.error(str(exc)) return st.caption(f'{start:%Y-%m-%d %H:%M} to {end:%Y-%m-%d %H:%M} UTC') summary = period_summary_df(conn, start, end) summary_row = summary.iloc[0] if not summary.empty else {} _metric_grid([ ('Scans', _integer(summary_row.get('scans', 0))), ('Findings', _integer(summary_row.get('findings', 0))), ('Errors', _integer(summary_row.get('errors', 0))), ('Checks', _integer(summary_row.get('checks', 0))), ('New targets', _integer(summary_row.get('queued_new', 0))), ('Updated rescans', _integer(summary_row.get('queued_updated', 0))), ('New alive', _integer(summary_row.get('new_alive', 0))), ('New usable', _integer(summary_row.get('new_usable', 0))), ]) _metric_grid([ ('Alive now', _integer(summary_row.get('alive_now', 0))), ('Usable LLM now', _integer(summary_row.get('usable_llm_now', 0))), ('Target queue now', _integer(summary_row.get('target_backlog', 0))), ('Keycheck queue now', _integer(summary_row.get('keycheck_backlog', 0))), ]) source_activity = source_activity_df(conn, start, end) new_alive = new_alive_breakdown_df(conn, start, end) source_col, alive_col = st.columns([1.2, 1], gap='large') with source_col: st.markdown('**Sources in period**') if not source_activity.empty: source_activity = source_activity.copy() source_activity['findings / scan'] = source_activity.apply( lambda row: round(_integer(row.get('findings')) / max(1, _integer(row.get('scans'))), 2), axis=1, ) display_df(source_activity, height=340) with alive_col: st.markdown('**Alive discovered in period**') display_df(new_alive, height=340) st.divider() st.subheader('Runtime now') st.caption('Current state is not restricted by the reporting window.') slots = scan_slots_df(scan_limiter_db) pipeline = pipeline_snapshot_df(conn) pipeline_row = pipeline.iloc[0] if not pipeline.empty else {} drive = os.path.splitdrive(work_dir or '')[0] or 'Scratch' _metric_grid([ ('Scan slots', f"{len(slots)}/{max_active_scans or '?'}"), (f'{drive} free', f'{disk_free_gb(work_dir):.2f} GiB'), ('Bundle backlog', _integer(pipeline_row.get('bundle_items', 0))), ('Projection lag', _integer(pipeline_row.get('projection_items', 0))), ('Candidate capacity', _integer(pipeline_row.get('keycheck_items', 0))), ('Quarantine capacity', _integer(pipeline_row.get('quarantine_items', 0))), ]) runtime, _ = parse_supervisor_status(log_dir) if not runtime.empty: runtime = runtime[runtime['source'].isin(CORE_RUNTIME_SOURCES)].copy() order = {name: index for index, name in enumerate([ 'result-ingester', 'jsonl-projector', 'janitor', 'worker-api', 'github', 'gitlab', 'huggingface', 'dockerhub', 'package_git', 'keychecks', ])} runtime['_order'] = runtime['source'].map(order).fillna(len(order)) runtime = runtime.sort_values('_order').drop(columns=['_order']) runtime_columns = ['source', 'runtime_status', 'up', 'next', 'restarts'] display_df(runtime[[column for column in runtime_columns if column in runtime]], height=360) else: st.info('Supervisor status is not available yet.') def page_overview(conn, results_dir, queue_dir, log_dir, work_dir, keycheck_dir, scan_limiter_db, max_active_scans): st.header('Overview') today_expr = today_sql() totals = query_df(conn, ''' SELECT COUNT(*) AS cycles, COALESCE(SUM(scanned_count), 0) AS scanned, COALESCE(SUM(found_count), 0) AS found_targets, COALESCE(SUM(error_count), 0) AS error_targets, COALESCE(SUM(skipped_count), 0) AS skipped, COALESCE(SUM(findings_count), 0) AS findings FROM source_cycles ''') finding_totals = query_df(conn, 'SELECT COUNT(*) AS finding_rows FROM findings') today = query_df(conn, ''' SELECT COALESCE(SUM(scanned_count), 0) AS scanned_today, COALESCE(SUM(findings_count), 0) AS findings_today, COALESCE(SUM(error_count), 0) AS errors_today FROM source_cycles WHERE started_at >= {today_expr} '''.format(today_expr=today_expr)) runtime_health, status_path = runtime_source_health(conn, log_dir) active_sources = int(runtime_health['runtime_status'].astype(str).str.lower().eq('running').sum()) if not runtime_health.empty and 'runtime_status' in runtime_health else 0 slots = scan_slots_df(scan_limiter_db) validation_available = table_exists(conn, 'keycheck_current_state') load_unique_totals = st.checkbox('Load unique scanner totals (slower)', value=False) load_alive_total = st.checkbox('Load all-time alive key count (slower)', value=False) if load_unique_totals: unique_totals = query_df(conn, ''' SELECT COUNT(DISTINCT NULLIF(secret_hash, '')) AS unique_secrets, COUNT(DISTINCT NULLIF(finding_fingerprint, '')) AS unique_findings FROM findings ''') urow = unique_totals.iloc[0] if not unique_totals.empty else {} unique_secrets = int(urow.get('unique_secrets', 0)) unique_findings = int(urow.get('unique_findings', 0)) else: unique_secrets = 'off' unique_findings = 'off' if validation_available: alive_total = scalar(conn, "SELECT COUNT(*) FROM keycheck_current_state WHERE status_group = 'alive'") if load_alive_total else 'off' alive_today = scalar(conn, f"SELECT COUNT(*) FROM keycheck_current_state WHERE status_group = 'alive' AND checked_at >= {today_expr}") else: alive_total = 0 alive_today = 0 row = totals.iloc[0] if not totals.empty else {} frow = finding_totals.iloc[0] if not finding_totals.empty else {} trow = today.iloc[0] if not today.empty else {} authoritative_queue_backlog = ( scalar(conn, "SELECT COUNT(*) FROM target_queue WHERE status IN ('pending','deferred')") if table_exists(conn, 'target_queue') else queue_backlog(queue_dir) ) show_metrics([ ('Active sources', active_sources), ('Scan slots', f"{len(slots)}/{max_active_scans or '?'}"), ('Queue backlog', int(authoritative_queue_backlog or 0)), ('Scanned today', int(trow.get('scanned_today', 0))), ('Findings today', int(trow.get('findings_today', 0))), ('Alive keys total', alive_total if isinstance(alive_total, str) else int(alive_total or 0)), ('Alive rows today', int(alive_today or 0)), ('Errors today', int(trow.get('errors_today', 0))), ]) if table_exists(conn, 'pipeline_capacity'): pipeline = query_df(conn, 'SELECT * FROM pipeline_capacity WHERE id = 1') prow = pipeline.iloc[0] if not pipeline.empty else {} show_metrics([ ('Bundle backlog', int(prow.get('bundle_items', 0))), ('Projection lag', int(prow.get('projection_items', 0))), ('Keycheck candidates', int(prow.get('keycheck_items', 0))), ('Pipeline quarantine', int(prow.get('quarantine_items', 0))), ]) if table_exists(conn, 'pipeline_leases'): leases = query_df(conn, ''' SELECT worker_name, state, heartbeat_at, lease_expires_at, last_error FROM pipeline_leases ORDER BY worker_name ''') st.caption('Pipeline worker leases') display_df(leases, height=150) st.caption(f"Scanner totals: cycles={int(row.get('cycles', 0))}, scanned={int(row.get('scanned', 0))}, findings={int(frow.get('finding_rows', 0))}, unique secrets={unique_secrets}, unique findings={unique_findings}. D free: {disk_free_gb(work_dir)} GB") archive_summary, archive_detectors, archive_targets = github_archive_yield(conn) if archive_summary: st.subheader('GitHub Archive Yield') show_metrics([ ('Archive found targets', archive_summary['found_targets']), ('Archive raw findings', archive_summary['raw_findings']), ('Archive interesting', archive_summary['interesting_findings']), ('Archive noise/generic', archive_summary['noise_or_generic_findings']), ('Archive verified', archive_summary['verified_findings']), ]) col_archive_1, col_archive_2 = st.columns(2) with col_archive_1: st.caption('Detector split from redacted findings metadata; raw result payloads are not loaded.') display_df(archive_detectors, height=320) with col_archive_2: st.caption('Targets ranked by interesting detector rows.') display_df(archive_targets, height=320) st.subheader('Keycheck PostgreSQL Current State And Compatibility Lag') summary_df, summary_path = keycheck_summary_df(keycheck_dir) db_summary = recent_keycheck_db_summary(conn) metric_summary = keycheck_db_metric_summary(keycheck_dir) if not summary_df.empty: merged = summary_df.merge(db_summary, on='service', how='left') if not db_summary.empty else summary_df if not metric_summary.empty: merged = merged.merge(metric_summary, on='service', how='left') preferred = [ 'service', 'alive', 'alive_rate_limited', 'no_balance', 'no_quota', 'limited', 'network', 'dead', 'restricted', 'file_updated_at', 'db_latest_checked', 'db_latest_created', 'db_recent_rows', 'db_cached_rows', 'db_write_failed_tail', 'cached_occurrence_tail', 'output_dir', ] display_df(merged[[column for column in preferred if column in merged.columns]], height=420) st.caption(f'Current-state summary: {summary_path}') else: st.info('No keycheck summary.tsv found') health_df, health_path = keycheck_pipeline_health(log_dir) st.subheader('Keycheck Pipeline Health') st.caption(health_path) display_df(health_df, height=160) if st.checkbox('Load keycheckable/noise totals (slower)', value=False): classified_totals = query_df(conn, f''' SELECT COUNT(*) AS raw_findings, COUNT(DISTINCT NULLIF(secret_hash, '')) AS raw_unique, COUNT(DISTINCT CASE WHEN is_keycheckable = 1 THEN NULLIF(secret_hash, '') END) AS keycheckable_unique, COUNT(DISTINCT CASE WHEN is_keycheckable = 0 THEN NULLIF(secret_hash, '') END) AS noise_unique FROM ({keycheckable_findings_view_sql()}) ''') crow = classified_totals.iloc[0] if not classified_totals.empty else {} show_metrics([ ('Raw unique findings', int(crow.get('raw_unique', 0))), ('Keycheckable unique', int(crow.get('keycheckable_unique', 0))), ('Noise unique', int(crow.get('noise_unique', 0))), ('Noise share', format_pct((int(crow.get('noise_unique', 0)) / int(crow.get('raw_unique', 1))) if int(crow.get('raw_unique', 0)) else 0)), ]) st.subheader('Runtime Source Health') st.caption(f'Runtime status file: {status_path}') display_df(runtime_health, height=420) st.subheader('Active Scan Slots') st.caption(scan_limiter_db) display_df(slots[['owner_source', 'owner_pid', 'age_sec', 'command']] if not slots.empty else slots, height=260) st.subheader('Per Source DB Aggregates') source_df = query_df(conn, ''' SELECT source, COUNT(*) AS cycles, SUM(fetched_count) AS fetched, SUM(queued_new_count) AS queued_new, SUM(queued_updated_count) AS queued_updated, SUM(scanned_count) AS scanned, SUM(found_count) AS found_targets, SUM(error_count) AS error_targets, SUM(skipped_count) AS skipped, SUM(findings_count) AS findings, SUM(verified_findings_count) AS verified_findings, AVG(hit_rate) AS avg_hit_rate, AVG(error_rate) AS avg_error_rate, AVG(targets_per_hour) AS avg_targets_per_hour, MAX(started_at) AS latest_cycle FROM source_cycles GROUP BY source ORDER BY latest_cycle DESC ''') if not source_df.empty: source_df['avg_hit_rate'] = source_df['avg_hit_rate'].map(format_pct) source_df['avg_error_rate'] = source_df['avg_error_rate'].map(format_pct) display_df(source_df) st.subheader('Current Queues') display_df(current_queue_counts(queue_dir)) colv1, colv2 = st.columns(2) with colv1: st.subheader('Top Detectors (Recent)') display_df(query_df(conn, ''' SELECT source, detector_name, COUNT(*) AS findings, COUNT(DISTINCT NULLIF(secret_hash, '')) AS unique_secrets FROM ( SELECT source, detector_name, secret_hash FROM findings ORDER BY id DESC LIMIT 20000 ) GROUP BY source, detector_name ORDER BY findings DESC LIMIT 20 '''), height=360) with colv2: st.subheader('Top Useful Sources By Alive Keys (Recent)') if validation_available: display_df(query_df(conn, ''' SELECT source, service, COUNT(DISTINCT key_hash) AS alive_keys, COUNT(*) AS linked_findings FROM ( SELECT source, service, key_hash, status_group FROM keycheck_results ORDER BY id DESC LIMIT 20000 ) WHERE status_group = 'alive' AND source IS NOT NULL AND key_hash != '' GROUP BY source, service ORDER BY alive_keys DESC, linked_findings DESC LIMIT 20 '''), height=360) else: st.info('No keycheck_results yet') col1, col2 = st.columns(2) with col1: st.subheader('Recent Source Cycles') display_df(query_df(conn, ''' SELECT started_at, ended_at, status, source, mode, query, fetched_count, queued_new_count, queued_updated_count, scanned_count, findings_count, error_count, skipped_count FROM source_cycles ORDER BY id DESC LIMIT 20 '''), height=420) with col2: st.subheader('Top Error Categories') errors = query_df(conn, ''' SELECT source, category, COUNT(*) AS count FROM errors GROUP BY source, category ORDER BY count DESC LIMIT 20 ''') display_df(errors, height=420) def page_runtime(conn, queue_dir, log_dir, work_dir, scan_limiter_db, max_active_scans): st.header('Runtime') runtime_health, status_path = runtime_source_health(conn, log_dir) slots = scan_slots_df(scan_limiter_db) active_sources = int(runtime_health['runtime_status'].astype(str).str.lower().eq('running').sum()) if not runtime_health.empty and 'runtime_status' in runtime_health else 0 waiting_sources = int(runtime_health['runtime_status'].astype(str).str.lower().isin(('waiting', 'blocked', 'backoff')).sum()) if not runtime_health.empty and 'runtime_status' in runtime_health else 0 show_metrics([ ('Running sources', active_sources), ('Waiting sources', waiting_sources), ('Active scan slots', f"{len(slots)}/{max_active_scans or '?'}"), ('Queue backlog', queue_backlog(queue_dir)), ('D/free GB', disk_free_gb(work_dir)), ]) st.subheader('Source Health') st.caption(f'Runtime status file: {status_path}') display_df(runtime_health, height=440) col1, col2 = st.columns(2) with col1: st.subheader('Active Scan Slots') st.caption(scan_limiter_db) display_df(slots[['owner_source', 'owner_pid', 'age_sec', 'command']] if not slots.empty else slots, height=360) with col2: st.subheader('Current Queues') display_df(current_queue_counts(queue_dir), height=360) st.subheader('Log Metadata') rows = [] if os.path.isdir(log_dir): for name in sorted(item for item in os.listdir(log_dir) if item.lower().endswith('.log')): path = os.path.join(log_dir, name) rows.append({ 'log': name, 'age': human_age(datetime.fromtimestamp(os.path.getmtime(path)).isoformat(timespec='seconds')) if os.path.exists(path) else '', 'bytes': os.path.getsize(path) if os.path.exists(path) else 0, }) display_df(pd.DataFrame(rows), height=360) def page_scanner_results(conn): st.header('Scanner Results') cycles = query_df(conn, ''' SELECT id AS cycle_id, started_at, ended_at, source, mode, query, status, fetched_count, queued_new_count, queued_updated_count, scanned_count, clean_count, found_count, skipped_count, error_count, findings_count, verified_findings_count, unique_secrets_count, unique_findings_count, hit_rate, error_rate, duration_sec FROM source_cycles ORDER BY id DESC LIMIT 5000 ''') if cycles.empty: st.info('No source cycle data yet') return col1, col2, col3, col4 = st.columns(4) with col1: source_filter = st.multiselect('Sources', sorted(cycles['source'].dropna().unique()), default=sorted(cycles['source'].dropna().unique())) with col2: status_filter = st.multiselect('Cycle statuses', sorted(cycles['status'].dropna().unique()), default=sorted(cycles['status'].dropna().unique())) with col3: query_contains = st.text_input('Query contains') with col4: days = st.number_input('Last N days (0 = all loaded)', min_value=0, max_value=365, value=0, step=1) filtered = cycles if source_filter: filtered = filtered[filtered['source'].isin(source_filter)] if status_filter: filtered = filtered[filtered['status'].isin(status_filter)] if query_contains: filtered = filtered[filtered['query'].astype(str).str.contains(query_contains, case=False, na=False)] if days: cutoff = pd.Timestamp.utcnow() - pd.Timedelta(days=int(days)) started = pd.to_datetime(filtered['started_at'], errors='coerce', utc=True) filtered = filtered[started >= cutoff] show_metrics([ ('Cycles', len(filtered)), ('Scanned', int(filtered['scanned_count'].fillna(0).sum())), ('Findings', int(filtered['findings_count'].fillna(0).sum())), ('Errors', int(filtered['error_count'].fillna(0).sum())), ('Skipped', int(filtered['skipped_count'].fillna(0).sum())), ('Hit rate', format_pct((filtered['found_count'].fillna(0).sum() / filtered['scanned_count'].fillna(0).sum()) if filtered['scanned_count'].fillna(0).sum() else 0)), ]) section = st.selectbox('Scanner results section', ['By Source', 'By Query', 'By Cycle', 'Detectors', 'Keycheckable vs Noise', 'Targets']) if section == 'By Source': by_source = filtered.groupby('source', dropna=False).agg( cycles=('cycle_id', 'count'), scanned=('scanned_count', 'sum'), findings=('findings_count', 'sum'), errors=('error_count', 'sum'), skipped=('skipped_count', 'sum'), ).reset_index().sort_values(['findings', 'scanned'], ascending=False) display_df(by_source, height=420) if not by_source.empty: st.plotly_chart(px.bar(by_source, x='source', y='findings', title='Findings By Source'), width='stretch') elif section == 'By Query': by_query = filtered.groupby(['source', 'query'], dropna=False).agg( cycles=('cycle_id', 'count'), scanned=('scanned_count', 'sum'), findings=('findings_count', 'sum'), errors=('error_count', 'sum'), ).reset_index().sort_values(['findings', 'scanned'], ascending=False) display_df(by_query, height=520) elif section == 'By Cycle': display_df(filtered.sort_values('started_at', ascending=False), height=620) chart_df = filtered.sort_values('started_at') if not chart_df.empty: st.plotly_chart(px.line(chart_df, x='started_at', y='findings_count', color='source', markers=True, title='Findings By Cycle'), width='stretch') elif section == 'Detectors': clauses = [] params = [] if source_filter: clauses.append('source IN ({})'.format(','.join('?' for _ in source_filter))) params.extend(source_filter) if query_contains: clauses.append('query LIKE ?') params.append(f'%{query_contains}%') where = ('WHERE ' + ' AND '.join(clauses)) if clauses else '' detectors = query_df(conn, f''' SELECT detector_name, source, validation_service, noise_reason, is_keycheckable, COUNT(*) AS findings, COUNT(DISTINCT NULLIF(secret_hash, '')) AS unique_secrets FROM ({keycheckable_findings_view_sql()}) {where} GROUP BY detector_name, source, validation_service, noise_reason, is_keycheckable ORDER BY findings DESC LIMIT 200 ''', params) display_df(detectors, height=620) elif section == 'Keycheckable vs Noise': st.warning('This diagnostic query scans findings and can be slow on the active DB.') if not st.button('Run keycheckable/noise query'): st.info('Press the button to calculate keycheckable/noise backlog.') return clauses = [] params = [] if source_filter: clauses.append('source IN ({})'.format(','.join('?' for _ in source_filter))) params.extend(source_filter) if query_contains: clauses.append('query LIKE ?') params.append(f'%{query_contains}%') where = ' AND '.join(clauses) backlog = query_df(conn, f''' SELECT source, query, validation_service, detector_name, SUM(raw_findings) AS raw_findings, SUM(unique_secrets) AS unique_secrets, SUM(keycheckable_unique) AS keycheckable_unique, SUM(noise_unique) AS noise_unique, MAX(checked_unique) AS checked_unique, MAX(alive_unique) AS alive_unique, MAX(pending_unique) AS pending_unique FROM ({keycheckable_backlog_sql(where)}) GROUP BY source, query, validation_service, detector_name ORDER BY keycheckable_unique DESC, noise_unique DESC, raw_findings DESC LIMIT 500 ''', params) display_df(backlog, height=620) elif section == 'Targets': targets = query_df(conn, ''' SELECT ended_at, source, query, status, target, normalized_target, duration_sec, findings_count, verified_findings_count, error_count, package_name, package_version, package_filename, package_type, package_size FROM target_scans ORDER BY id DESC LIMIT 5000 ''') if not targets.empty: if source_filter: targets = targets[targets['source'].isin(source_filter)] if query_contains: targets = targets[targets['query'].astype(str).str.contains(query_contains, case=False, na=False)] display_df(targets, height=620) def page_runs(conn): st.header('Historical Runs (Legacy / Debug)') st.caption('Loop-mode sources usually do not finish runs, so these totals are not the source of truth for scanner metrics.') runs = query_df(conn, ''' SELECT id, started_at, ended_at, duration_sec, status, invocation_mode, selected_source, selected_platform, config_path, total_fetched, total_queued_new, total_scanned, total_clean, total_found, total_skipped, total_errors, total_findings, total_verified_findings, total_unique_secrets, total_unique_findings FROM runs ORDER BY id DESC ''') display_df(runs) if not runs.empty: run_id = st.selectbox('Inspect run', runs['id'].tolist()) config = query_df(conn, 'SELECT scope, source, captured_at FROM config_snapshots WHERE run_id = ? ORDER BY id', [int(run_id)]) display_df(config) def page_sources(conn): st.header('Sources') cycles = query_df(conn, ''' SELECT started_at, source, mode, query, status, fetched_count, queued_new_count, queued_updated_count, scanned_count, clean_count, found_count, skipped_count, error_count, findings_count, verified_findings_count, unique_secrets_count, unique_findings_count, targets_per_hour, hit_rate, error_rate FROM source_cycles ORDER BY id DESC ''') if cycles.empty: st.info('No source cycle data yet') return source_filter = st.multiselect('Sources', sorted(cycles['source'].dropna().unique()), default=sorted(cycles['source'].dropna().unique())) filtered = cycles[cycles['source'].isin(source_filter)] if source_filter else cycles display_df(filtered) chart_df = filtered.sort_values('started_at') if not chart_df.empty: st.plotly_chart(px.line(chart_df, x='started_at', y='scanned_count', color='source', markers=True, title='Scanned targets by cycle'), width='stretch') st.plotly_chart(px.line(chart_df, x='started_at', y='error_rate', color='source', markers=True, title='Error rate by cycle'), width='stretch') def page_queries(conn): st.header('Queries') queries = query_df(conn, ''' SELECT source, query, COUNT(*) AS cycles, SUM(fetched_count) AS fetched, SUM(queued_new_count) AS queued_new, SUM(queued_updated_count) AS queued_updated, SUM(scanned_count) AS scanned, SUM(findings_count) AS findings, SUM(error_count) AS errors, AVG(hit_rate) AS avg_hit_rate, AVG(error_rate) AS avg_error_rate, AVG(duration_sec) AS avg_duration_sec FROM source_cycles GROUP BY source, query ORDER BY findings DESC, scanned DESC ''') if not queries.empty: queries['avg_hit_rate'] = queries['avg_hit_rate'].map(format_pct) queries['avg_error_rate'] = queries['avg_error_rate'].map(format_pct) display_df(queries) def page_findings(conn): st.header('Findings') st.caption('Redacted scanner metadata. Raw secret columns are never queried by this dashboard.') desired_columns = [ 'id', 'created_at', 'source', 'query', 'target', 'detector_name', 'detector_type', 'verified', 'provider', 'credential_kind', 'credential_confidence', 'required_context_missing', 'principal', 'username', 'email', 'project_id', 'organization', 'registry', 'endpoint', 'scope', 'resource', 'file_path', 'line_number', 'commit_hash', 'secret_hash', 'detector_secret_hash', 'finding_fingerprint', 'redacted_secret', ] existing = table_columns(conn, 'findings') columns = ', '.join(column for column in desired_columns if column in existing) if not columns: st.info('No findings columns available') return findings = query_df(conn, f'SELECT {columns} FROM findings ORDER BY id DESC LIMIT 1000') if findings.empty: st.info('No findings yet') return col1, col2, col3, col4 = st.columns(4) col1.metric('Rows shown', len(findings)) col2.metric('Unique secrets', findings['secret_hash'].replace('', pd.NA).dropna().nunique()) col3.metric('Unique findings', findings['finding_fingerprint'].replace('', pd.NA).dropna().nunique()) col4.metric('Verified', int(findings['verified'].sum())) detectors = query_df(conn, 'SELECT detector_name, COUNT(*) AS count FROM findings GROUP BY detector_name ORDER BY count DESC LIMIT 30') if not detectors.empty: st.plotly_chart(px.bar(detectors, x='detector_name', y='count', title='Findings by detector'), width='stretch') display_df(findings) def page_validation(conn): st.header('Validation / Keychecks') if not table_exists(conn, 'keycheck_results'): st.info('No keycheck_results table yet. New keycheck runs will create it and populate validation data.') return preset = st.selectbox('Preset', VALIDATION_PRESETS, index=0) historical_preset = preset in ('Ever usable LLM keys', 'Ever alive / LLM candidates') preset_uses_latest = preset != 'Latest rows' and not historical_preset if preset == 'Unattributed alive': default_status_groups = ['alive'] default_access_tiers = ['alive_unproven_llm', 'usable_llm'] elif preset == 'Latest rows': default_status_groups = ['alive'] default_access_tiers = VALIDATION_ACCESS_TIERS elif preset == 'Usable LLM keys': default_status_groups = VALIDATION_STATUS_GROUPS default_access_tiers = ['usable_llm'] elif preset == 'Ever usable LLM keys': default_status_groups = VALIDATION_STATUS_GROUPS default_access_tiers = ['usable_llm'] elif preset == 'Ever alive / LLM candidates': default_status_groups = ['alive'] default_access_tiers = ['usable_llm', 'alive_unproven_llm'] elif preset == 'Quota / no balance': default_status_groups = VALIDATION_STATUS_GROUPS default_access_tiers = ['no_quota', 'quota_limited'] elif preset == 'Alive but not proven LLM': default_status_groups = VALIDATION_STATUS_GROUPS default_access_tiers = ['alive_unproven_llm'] else: default_status_groups = VALIDATION_STATUS_GROUPS default_access_tiers = VALIDATION_ACCESS_TIERS col1, col2, col3, col4 = st.columns(4) with col1: services = st.multiselect('Services', VALIDATION_SERVICES, default=VALIDATION_SERVICES) with col2: status_groups = st.multiselect('Status groups', VALIDATION_STATUS_GROUPS, default=default_status_groups) source_options = [UNATTRIBUTED, *SOURCES] with col3: sources = st.multiselect('Sources', source_options, default=source_options) with col4: row_limit = st.number_input('Rows limit', min_value=500, max_value=50000, value=5000, step=500) col5, col6, col7 = st.columns([1, 1, 2]) with col5: checked_axis = st.selectbox('Time axis', ['checked_at', 'found_at']) with col6: dedupe_latest = st.checkbox('Dedupe latest per key (slower)', value=preset_uses_latest) with col7: query_text = st.text_input('Source query contains') access_tiers = st.multiselect('Access tiers', VALIDATION_ACCESS_TIERS, default=default_access_tiers) exact_statuses = st.multiselect('Exact provider statuses', VALIDATION_STATUSES, default=[]) search_text = st.text_input( 'Search hash / masked key / target', help='Use a SHA-256 hash, masked value, or non-secret metadata. Raw credentials are not accepted or queried.', ) if search_text and not dashboard_search_is_safe(search_text): st.warning('Search refused: use a SHA-256 hash or masked/non-secret metadata.') search_text = '' if historical_preset: st.caption('Historical preset: uses all matching keycheck rows, not latest current-state. This answers “which source ever produced this alive/usable key”.') search_like = f'%{search_text}%' if search_text else '' search_hash = search_text.lower() if re.fullmatch(r'[0-9a-fA-F]{64}', search_text or '') else '' clauses = [] params = [] id_window = None if not dedupe_latest and not search_text and not historical_preset: max_id_df = query_df(conn, 'SELECT MAX(id) AS max_id FROM keycheck_results') max_id = int(max_id_df.iloc[0]['max_id'] or 0) if not max_id_df.empty else 0 id_window = max(0, max_id - int(row_limit) * 20) clauses.append('kr.id >= ?') params.append(id_window) if services and set(services) != set(VALIDATION_SERVICES): clauses.append('kr.service IN ({})'.format(','.join('?' for _ in services))) params.extend(services) if status_groups and set(status_groups) != set(VALIDATION_STATUS_GROUPS): clauses.append('kr.status_group IN ({})'.format(','.join('?' for _ in status_groups))) params.extend(status_groups) if sources and set(sources) != set(source_options): source_clauses = [] concrete_sources = [item for item in sources if item != UNATTRIBUTED] if concrete_sources: source_clauses.append('COALESCE(kr.source, f.source, fh.source) IN ({})'.format(','.join('?' for _ in concrete_sources))) params.extend(concrete_sources) if UNATTRIBUTED in sources: source_clauses.append('COALESCE(kr.source, f.source, fh.source) IS NULL') if source_clauses: clauses.append('(' + ' OR '.join(source_clauses) + ')') if query_text: clauses.append("COALESCE(kr.query, f.query, fh.query, '') LIKE ?") params.append(f'%{query_text}%') access_tier_sql = validation_access_tier_sql('kr') if access_tiers and set(access_tiers) != set(VALIDATION_ACCESS_TIERS): clauses.append(f"({access_tier_sql}) IN ({','.join('?' for _ in access_tiers)})") params.extend(access_tiers) if exact_statuses: clauses.append('UPPER(kr.status) IN ({})'.format(','.join('?' for _ in exact_statuses))) params.extend(exact_statuses) if preset == 'Unattributed alive': clauses.append("kr.status_group = 'alive'") clauses.append("COALESCE(kr.source, kr.finding_id, kr.target_scan_id) IS NULL") if search_text: clauses.append('''( kr.key_masked LIKE ? OR kr.key_hash = ? OR kr.secret_hash = ? OR kr.detector_name LIKE ? OR kr.target LIKE ? OR COALESCE(kr.source, f.source, fh.source, '') LIKE ? OR COALESCE(kr.query, f.query, fh.query, '') LIKE ? OR COALESCE(kr.target, f.target, fh.target, ts.target, '') LIKE ? OR COALESCE(f.redacted_secret, fh.redacted_secret, '') LIKE ? OR COALESCE(f.detector_name, fh.detector_name, '') LIKE ? OR COALESCE(f.file_path, fh.file_path, '') LIKE ? OR COALESCE(f.commit_hash, fh.commit_hash, '') LIKE ? OR ts.package_name LIKE ? OR ts.package_version LIKE ? OR ts.package_filename LIKE ? )''') params.extend([ search_like, search_hash, search_hash, search_like, search_like, search_like, search_like, search_like, search_like, search_like, search_like, search_like, search_like, search_like, search_like, ]) where = ('WHERE ' + ' AND '.join(clauses)) if clauses else '' if dedupe_latest: source_sql = f'({latest_keycheck_view_sql()})' else: source_sql = 'keycheck_results' meta_source = json_extract_sql('kr.metadata_json', '$.source') meta_backfill_source = json_extract_sql('kr.metadata_json', '$.backfill_source_file') meta_result_source = json_extract_sql('kr.metadata_json', '$.result_source') meta_llm_probe_status = json_extract_sql('kr.metadata_json', '$.llm_probe_status') meta_probe_status = json_extract_sql('kr.metadata_json', '$.probe.status') meta_llm_probe_model = json_extract_sql('kr.metadata_json', '$.llm_probe_model') meta_probe_model = json_extract_sql('kr.metadata_json', '$.probe.model') meta_remaining_credits = json_extract_sql('kr.metadata_json', '$.remaining_credits') meta_balance_usd = json_extract_sql('kr.metadata_json', '$.balance_usd') meta_vertex_enabled = json_extract_sql('kr.metadata_json', '$.vertex_enabled') meta_bedrock_enabled = json_extract_sql('kr.metadata_json', '$.bedrock_enabled') meta_route_probe = json_extract_sql('kr.metadata_json', '$.route_probe') meta_foundry_route_probe = json_extract_sql('kr.metadata_json', '$.foundry_route_probe') rows = query_df(conn, f''' SELECT kr.id, kr.checked_at, kr.found_at, kr.service, kr.status_group, kr.status, {validation_access_tier_sql('kr')} AS access_tier, kr.key_hash, kr.secret_hash, kr.key_masked, COALESCE(kr.source, f.source, fh.source) AS source, COALESCE(kr.query, f.query, fh.query) AS query, COALESCE(kr.cycle_id, f.cycle_id, fh.cycle_id) AS cycle_id, COALESCE(kr.target_scan_id, f.target_scan_id, fh.target_scan_id) AS target_scan_id, COALESCE(kr.finding_id, fh.id) AS finding_id, COALESCE(kr.detector_name, f.detector_name, fh.detector_name) AS detector_name, COALESCE(kr.target, f.target, fh.target, ts.target) AS target, COALESCE(f.file_path, fh.file_path) AS file_path, COALESCE(f.line_number, fh.line_number) AS line_number, COALESCE(f.commit_hash, fh.commit_hash) AS commit_hash, ts.package_name, ts.package_version, ts.package_filename, ts.package_type, CASE WHEN f.id IS NOT NULL THEN 'finding_id' WHEN fh.id IS NOT NULL THEN 'secret_hash' WHEN COALESCE(kr.source, kr.query, kr.target) IS NOT NULL THEN 'keycheck_metadata' ELSE 'missing_finding' END AS attribution_status, {meta_source} AS keycheck_source_line, {meta_backfill_source} AS backfill_source_file, COALESCE({meta_result_source}, 'api_check') AS result_source, COALESCE({meta_llm_probe_status}, {meta_probe_status}) AS llm_probe_status, COALESCE({meta_llm_probe_model}, {meta_probe_model}) AS llm_probe_model, {meta_remaining_credits} AS remaining_credits, {meta_balance_usd} AS balance_usd, {meta_vertex_enabled} AS vertex_enabled, {meta_bedrock_enabled} AS bedrock_enabled, {meta_route_probe} AS azure_route_probe, {meta_foundry_route_probe} AS foundry_route_probe FROM {source_sql} kr LEFT JOIN findings f ON f.id = kr.finding_id LEFT JOIN findings fh ON f.id IS NULL AND fh.id = ( SELECT id FROM findings WHERE secret_hash = COALESCE(NULLIF(kr.secret_hash, ''), NULLIF(kr.key_hash, '')) ORDER BY id DESC LIMIT 1 ) LEFT JOIN target_scans ts ON ts.id = COALESCE(kr.target_scan_id, f.target_scan_id, fh.target_scan_id) {where} ORDER BY kr.id DESC LIMIT ? ''', [*params, int(row_limit)]) if rows.empty: st.info('No validation rows match current filters.') if search_text: st.subheader('Finding Metadata Matches') st.caption('Fallback hash/redacted/metadata search in findings.') display_df(finding_metadata_search(conn, search_text), height=520) return filtered = rows.copy() if 'source' in filtered: filtered['source'] = filtered['source'].fillna(UNATTRIBUTED) if 'query' in filtered: filtered['query'] = filtered['query'].fillna(UNATTRIBUTED) if 'key_hash' in filtered: filtered['key_identity'] = filtered['key_hash'].where(filtered['key_hash'].astype(str) != '', filtered['secret_hash']).fillna(filtered['key_masked']) else: filtered['key_identity'] = filtered.get('key_masked', pd.Series(dtype='object')) usable_keys = int(filtered.loc[filtered['access_tier'] == 'usable_llm', 'key_identity'].replace('', pd.NA).dropna().nunique()) if 'access_tier' in filtered else 0 unproven_keys = int(filtered.loc[filtered['access_tier'] == 'alive_unproven_llm', 'key_identity'].replace('', pd.NA).dropna().nunique()) if 'access_tier' in filtered else 0 quota_keys = int(filtered.loc[filtered['access_tier'].isin(['no_quota', 'quota_limited']), 'key_identity'].replace('', pd.NA).dropna().nunique()) if 'access_tier' in filtered else 0 unattributed_rows = int((filtered['attribution_status'] == 'missing_finding').sum()) if 'attribution_status' in filtered else 0 show_metrics([ ('Rows', len(filtered)), ('Usable LLM keys', usable_keys), ('Alive unproven keys', unproven_keys), ('No quota / limited keys', quota_keys), ('Unattributed rows', unattributed_rows), ]) if not filtered.empty: status_by_service = filtered.groupby(['service', 'access_tier']).size().reset_index(name='count') st.plotly_chart(px.bar(status_by_service, x='service', y='count', color='access_tier', title='Validation Access Tier By Service'), width='stretch') usable_rows = filtered[filtered['access_tier'] == 'usable_llm'].copy() usable_rows['source'] = usable_rows['source'].fillna(UNATTRIBUTED) usable_rows['query'] = usable_rows['query'].fillna(UNATTRIBUTED) usable_rows['target'] = usable_rows['target'].fillna(UNATTRIBUTED) usable = usable_rows.groupby(['source', 'query', 'target', 'service', 'status', 'attribution_status'], dropna=False).agg( rows=('id', 'count'), keys=('key_identity', 'nunique'), latest_checked=('checked_at', 'max'), ).reset_index().sort_values(['keys', 'rows'], ascending=False).head(100) st.subheader('Usable LLM By Source / Query / Target') display_df(usable, height=420) unproven_rows = filtered[filtered['access_tier'] == 'alive_unproven_llm'].copy() if not unproven_rows.empty: unproven_rows['source'] = unproven_rows['source'].fillna(UNATTRIBUTED) unproven = unproven_rows.groupby(['source', 'query', 'service', 'status'], dropna=False).agg( rows=('id', 'count'), keys=('key_identity', 'nunique'), latest_checked=('checked_at', 'max'), ).reset_index().sort_values(['keys', 'rows'], ascending=False).head(100) st.subheader('Alive But Not Proven LLM') display_df(unproven, height=300) st.subheader('Usable LLM Origins') origin_columns = [ 'checked_at', 'found_at', 'service', 'status', 'access_tier', 'result_source', 'source', 'query', 'target', 'attribution_status', 'llm_probe_status', 'llm_probe_model', 'remaining_credits', 'balance_usd', 'vertex_enabled', 'bedrock_enabled', 'azure_route_probe', 'foundry_route_probe', 'keycheck_source_line', 'backfill_source_file', 'detector_name', 'file_path', 'line_number', 'commit_hash', 'package_name', 'package_version', 'package_filename', 'package_type', 'finding_id', 'target_scan_id', 'cycle_id', 'id', ] display_df( usable_rows[[column for column in origin_columns if column in usable_rows.columns]].head(int(row_limit)), height=520, ) if search_text: st.subheader('Finding Metadata Matches') st.caption('Direct hash/redacted/metadata matches, including rows without keycheck results.') display_df(finding_metadata_search(conn, search_text), height=420) cycle = filtered.groupby(['source', 'query', 'cycle_id', 'service', 'status_group']).size().reset_index(name='count').sort_values('count', ascending=False).head(200) st.subheader('Validation By Found Cycle') display_df(cycle, height=420) timeline = filtered.dropna(subset=[checked_axis]).copy() if not timeline.empty: timeline['day'] = timeline[checked_axis].astype(str).str.slice(0, 10) timeline_df = timeline.groupby(['day', 'status_group']).size().reset_index(name='count') st.plotly_chart(px.line(timeline_df, x='day', y='count', color='status_group', markers=True, title=f'Validation Timeline By {checked_axis}'), width='stretch') st.subheader('Keycheckable Backlog By Source / Query') st.caption('Optional diagnostic. Runs a heavier query over findings and keycheck_results.') if st.button('Calculate keycheckable backlog'): clauses = ['is_keycheckable = 1'] params = [] if services: clauses.append('validation_service IN ({})'.format(','.join('?' for _ in services))) params.extend(services) concrete_sources = [item for item in sources if item != UNATTRIBUTED] if concrete_sources: clauses.append('source IN ({})'.format(','.join('?' for _ in concrete_sources))) params.extend(concrete_sources) if query_text: clauses.append('query LIKE ?') params.append(f'%{query_text}%') backlog = query_df(conn, f''' SELECT source, query, validation_service, SUM(raw_findings) AS raw_findings, SUM(unique_secrets) AS unique_secrets, SUM(keycheckable_unique) AS keycheckable_unique, MAX(checked_unique) AS checked_unique, MAX(alive_unique) AS alive_unique, MAX(pending_unique) AS pending_unique FROM ({keycheckable_backlog_sql(' AND '.join(clauses))}) GROUP BY source, query, validation_service ORDER BY pending_unique DESC, keycheckable_unique DESC, alive_unique DESC LIMIT 200 ''', params) display_df(backlog, height=420) st.subheader('Latest Validation Rows') display_df(filtered, height=620) def page_errors(conn): st.header('Errors') grouped = query_df(conn, ''' SELECT source, category, COUNT(*) AS count, MAX(created_at) AS latest FROM errors GROUP BY source, category ORDER BY count DESC, latest DESC LIMIT 200 ''') display_df(grouped) recent = query_df(conn, ''' SELECT created_at, source, query, target, category FROM errors ORDER BY id DESC LIMIT 300 ''') st.subheader('Recent Errors') display_df(recent) def page_targets(conn): st.header('Targets') targets = query_df(conn, ''' SELECT ended_at, source, query, status, target, normalized_target, duration_sec, findings_count, verified_findings_count, error_count, package_name, package_version, package_filename, package_type, package_size FROM target_scans ORDER BY id DESC LIMIT 2000 ''') if targets.empty: st.info('No target scan data yet') return statuses = st.multiselect('Statuses', sorted(targets['status'].dropna().unique()), default=sorted(targets['status'].dropna().unique())) source_values = st.multiselect('Sources', sorted(targets['source'].dropna().unique()), default=sorted(targets['source'].dropna().unique())) text = st.text_input('Target contains') filtered = targets if statuses: filtered = filtered[filtered['status'].isin(statuses)] if source_values: filtered = filtered[filtered['source'].isin(source_values)] if text: filtered = filtered[filtered['target'].str.contains(text, case=False, na=False)] display_df(filtered) def page_queues_state(conn, queue_dir, state_file): st.header('Queues And State') st.subheader('Current Queue Files') display_df(current_queue_counts(queue_dir)) st.subheader('Queue Snapshots') snapshots = query_df(conn, ''' SELECT captured_at, source, phase, todo_count, checked_count, todo_file, checked_file FROM queue_snapshots ORDER BY captured_at DESC LIMIT 1000 ''') display_df(snapshots) state, path = load_runner_state(state_file) st.subheader('Runner State') st.caption(path) if state: state_rows = [] for source, item in (state.get('sources') or {}).items(): state_rows.append({ 'source': source, 'query_index': item.get('query_index'), 'last_query': item.get('last_query'), 'last_auth': item.get('last_auth'), 'last_status': item.get('last_status'), 'cycles': item.get('cycles'), 'last_scanned': item.get('last_scanned'), }) display_df(pd.DataFrame(state_rows)) else: st.info('runner_state.json not found or empty') def page_config(conn): st.header('Config Snapshots') st.caption('Payloads are unavailable because historical snapshots may contain credentials.') configs = query_df(conn, ''' SELECT captured_at, run_id, cycle_id, scope, source FROM config_snapshots ORDER BY captured_at DESC, id DESC LIMIT 500 ''') display_df(configs) def page_logs(log_dir): st.header('Logs') st.caption('Only file metadata is exposed. Log contents are never rendered or downloaded.') if not os.path.isdir(log_dir): st.info('Log directory not found') return log_files = sorted(name for name in os.listdir(log_dir) if name.lower().endswith('.log')) if not log_files: st.info('No .log files found') return rows = [] for name in log_files: path = os.path.join(log_dir, name) updated = datetime.fromtimestamp(os.path.getmtime(path)).isoformat(timespec='seconds') rows.append({ 'log': name, 'bytes': os.path.getsize(path), 'updated_at': updated, 'age': human_age(updated), }) display_df(pd.DataFrame(rows), height=520) def page_package_repos(conn): st.header('Package Git Candidates') candidates = query_df(conn, ''' SELECT last_seen_at, package_source, package_name, package_version, query, provider, repo_url, confidence FROM package_repo_candidates ORDER BY last_seen_at DESC LIMIT 2000 ''') display_df(candidates) def managed_config_argument(argv=None): values = list(sys.argv[1:] if argv is None else argv) for index, value in enumerate(values): if value == '--config' and index + 1 < len(values): return values[index + 1] if str(value).startswith('--config='): return str(value).split('=', 1)[1] return None def main(): try: require_active_supervisor_child( managed_config_argument(), child_kind='dashboard', require_dsn=True, ) dashboard_host = str(os.getenv('TRUF_DASHBOARD_HOST') or '') if os.getenv('TRUF_DASHBOARD_CANONICAL_LAUNCH') != '1' or not ipaddress.ip_address(dashboard_host).is_loopback: raise LifecycleAuthorityError('dashboard requires canonical loopback supervisor launch authority') except (LifecycleAuthorityError, ValueError) as exc: raise SystemExit(str(exc)) from exc args = parse_args() db_path = resolve_db_path(args) db_url = resolve_db_url(args) st.set_page_config( page_title='TRUF Status', page_icon='T', layout='wide', initial_sidebar_state='collapsed', ) try: conn = connect_db(db_path, db_url, args.immutable_db) except Exception as exc: _dashboard_styles() st.title('TRUF') st.warning(f'PostgreSQL observability is temporarily unavailable ({type(exc).__name__}). The dashboard will retry on refresh.') runtime_health, _ = parse_supervisor_status(args.log_dir) display_df(runtime_health, height=360) return if conn is None: _dashboard_styles() st.title('TRUF') st.info(f'No observability database found. Start a configured source through supervisor.py to create {DB_FILENAME}.') return try: page_simple_dashboard( conn, args.log_dir, args.work_dir, args.scan_limiter_db, args.max_active_scans, ) finally: try: conn.close() except Exception: pass if __name__ == '__main__': main()