"""Test dockerhub_eternal lane: offline fixture parsing + db_add + dedup.""" import hashlib import os import sqlite3 import sys import tempfile # Point DB to temp file os.environ["KF_DB"] = tempfile.mktemp(suffix=".db") # Import after setting env sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import dockerhub_eternal as dh # --- fixture: realistic config JSON with Env containing secrets --- FIXTURE_CONFIG = { "config": { "Env": [ "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", "OPENAI_API_KEY=sk-proj-AbCdEfGhIjKlMnOpQrStUvWxYz1234567890ABCDEFGHIJKLMNOPQRSTUVWXYZ", "BOT_TOKEN=1234567890:AAHjKlMnOpQrStUvWxYzAbCdEfGhIjKlMnO", "REDIS_URL=redis://:password@redis:6379", "PGPASS=postgresql://user:randompassword123@localhost:5432/db", "ANTHROPIC_KEY=sk-ant-api03-AbCdEfGhIjKlMnOpQrStUvWxYz1234567890ABCDEFGHIJKLMNOPQRSTUV-abCdEfGh", "PLACEHOLDER=sk-your-secret-key-here-1234567890", "TEST_VAR=test1234", "SHORT=abc", "DASHSCOPE_KEY=sk-aabbccddee0011223344556677889900", "HUGGINGFACE_TOKEN=hf_AbCdEfGhIjKlMnOpQrStUvWxYz1234567890AbCd", "ANY_SECRET=hf_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", ] } } DB_CONTRACT_SQL = """CREATE TABLE IF NOT EXISTS keys ( id INTEGER PRIMARY KEY AUTOINCREMENT, hash TEXT UNIQUE, val TEXT, prov TEXT, status TEXT DEFAULT "NEW", plan TEXT, price TEXT, remaining TEXT, found TEXT, repo TEXT)""" def setup_db(): conn = sqlite3.connect(os.environ["KF_DB"]) conn.execute(DB_CONTRACT_SQL) conn.commit() conn.close() def test_scan_env_whole_text(): """Config Env scanning finds keys and rejects junk.""" total_before = dh.total n = dh.scan_env_whole_text(FIXTURE_CONFIG, "docker:library/test:latest") assert n >= 4, f"Expected >=4 keys, got {n}" # Verify DB rows conn = sqlite3.connect(os.environ["KF_DB"]) rows = conn.execute("SELECT val, prov, found, repo FROM keys").fetchall() conn.close() vals = {r[0] for r in rows} provs = {r[1] for r in rows} # Should have OPENAI key assert any("sk-proj-AbCdEfGhIjKlMnOpQrStU" in v for v in vals), \ "Missing OpenAI key" assert "OPENAI" in provs, f"Missing OPENAI provider, got {provs}" # Should have TELEGRAM bot token assert any("AAHjKlMnOpQrStUvWxYz" in v for v in vals), \ "Missing Telegram token" assert "TELEGRAM" in provs, f"Missing TELEGRAM provider, got {provs}" # Should have ANTHROPIC key assert any("sk-ant-api03" in v for v in vals), \ "Missing Anthropic key" assert "ANTHROPIC" in provs, f"Missing ANTHROPIC provider, got {provs}" # Should have DASHSCOPE key assert any("sk-aabbccddee0011223344556677889900" in v for v in vals), \ "Missing DashScope key" assert "DASHSCOPE" in provs, f"Missing DASHSCOPE provider, got {provs}" assert "HUGGINGFACE" in provs, f"Missing HUGGINGFACE provider, got {provs}" # Junk filter should reject placeholder key for r in rows: assert "your" not in r[0].lower(), f"Junk leaked: {r[0]}" # Junk filter should reject test keys for r in rows: assert "test" not in r[0].lower() or "test" in r[2].lower(), \ f"Junk leaked: {r[0]}" print(f" Found {len(rows)} total keys, {len(set(vals))} unique") # Verify found tag for r in rows: assert r[3] == "docker:library/test:latest", f"Wrong repo: {r[3]}" assert r[2] == "dockerhub", f"Wrong found tag: {r[2]}" def test_dedup(): """Running scan_env_whole_text again should not duplicate (DB dedup).""" conn = sqlite3.connect(os.environ["KF_DB"]) before = conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0] conn.close() dh.scan_env_whole_text(FIXTURE_CONFIG, "docker:library/test:latest") conn = sqlite3.connect(os.environ["KF_DB"]) after = conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0] conn.close() assert after == before, f"Rows changed: {before} -> {after}" print(f" Dedup OK: {after} rows unchanged") def test_db_add_direct(): """db_add writes a row with correct schema.""" test_val = "sk-proj-AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" dh.db_add(test_val, "OPENAI", "docker:test/test:latest") conn = sqlite3.connect(os.environ["KF_DB"]) row = conn.execute( "SELECT val, prov, found, repo, hash FROM keys WHERE val=?", (test_val,) ).fetchone() conn.close() assert row is not None, "Row not found" assert row[2] == "dockerhub", f"Expected found='dockerhub', got '{row[2]}'" assert row[3] == "docker:test/test:latest" expected_hash = hashlib.sha1(row[0].encode()).hexdigest()[:16] assert row[4] == expected_hash, "Hash mismatch" print(f" db_add row: prov={row[1]} hash={row[4]}") def test_db_add_dedup(): """db_add with same key updates existing row (no duplicate).""" key = "sk-proj-BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB" dh.db_add(key, "OPENAI", "docker:test/a:latest") conn = sqlite3.connect(os.environ["KF_DB"]) count_before = conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0] conn.close() dh.db_add(key, "OPENAI", "docker:test/a:latest") conn = sqlite3.connect(os.environ["KF_DB"]) count_after = conn.execute("SELECT COUNT(*) FROM keys").fetchone()[0] conn.close() assert count_before == count_after, f"Dedup failed: {count_before} -> {count_after}" print(f" db_add dedup OK: {count_before} rows") def test_is_junk(): assert dh.is_junk("sk-your-secret"), "Did not catch 'your'" assert dh.is_junk("testtest12345"), "Did not catch 'testtest'" assert dh.is_junk("aaaaaa"), "Did not catch low-entropy" assert dh.is_junk("dummy_value_here"), "Did not catch 'DUMMY'" assert not dh.is_junk("sk-proj-AbCdEfGhIjKlMnOpQrStUvWxYz123456"), "False junk" print(" Junk filter OK") def test_scan_config_env(): """Legacy scan_config_env also works (non-ENV_RE patterns).""" n = dh.scan_config_env(FIXTURE_CONFIG, "docker:test/legacy:latest") # scan_config_env only processes individual env values as direct text patterns vals_sofar = set() conn = sqlite3.connect(os.environ["KF_DB"]) rows = conn.execute("SELECT val FROM keys").fetchall() conn.close() print(f" scan_config_env contributed {n} keys") if __name__ == "__main__": setup_db() test_is_junk() test_scan_env_whole_text() test_dedup() test_db_add_direct() test_db_add_dedup() test_scan_config_env() print("\nALL PASS")