153 lines
6.8 KiB
Python
153 lines
6.8 KiB
Python
"""Test wayback_eternal: config URL detection, junk filter, scan_text integration, db_add+dedup.
|
||
|
||
Sets KF_DB to a smoke DB before importing the module so the global conn
|
||
points to the test DB. Creates the keys table. No network calls.
|
||
"""
|
||
import hashlib, os, sqlite3, sys
|
||
|
||
# --- Point KF_DB to a smoke DB before module import ---
|
||
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
DB_PATH = os.path.join(THIS_DIR, "data", "_smoke_wayback.db")
|
||
os.environ["KF_DB"] = DB_PATH
|
||
|
||
os.makedirs(os.path.join(THIS_DIR, "data"), exist_ok=True)
|
||
|
||
# Drop + recreate the smoke DB so each test run is clean
|
||
if os.path.exists(DB_PATH):
|
||
os.remove(DB_PATH)
|
||
|
||
boot_conn = sqlite3.connect(DB_PATH)
|
||
boot_conn.execute("""
|
||
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)
|
||
""")
|
||
boot_conn.commit()
|
||
boot_conn.close()
|
||
|
||
# --- Now import the lane (reads KF_DB) ---
|
||
sys.path.insert(0, THIS_DIR)
|
||
import wayback_eternal as wb
|
||
|
||
|
||
# ── Tests ────────────────────────────────────────────────────────────
|
||
|
||
def test_is_config_url():
|
||
"""Proof: config URL detection matches known paths, rejects noise."""
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/.env")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/.env.example")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/.env.production")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/.env.local")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/credentials")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/credentials.json")
|
||
assert wb.is_config_url("https://gist.githubusercontent.com/user/123/raw/id_rsa")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/id_rsa.pub")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/.npmrc")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/application.properties")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/secret")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/secrets")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/.netrc")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/.git-credentials")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/.dockercfg")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/composer.json")
|
||
assert wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/credentials.yml")
|
||
|
||
# Should NOT match
|
||
assert not wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/index.js")
|
||
assert not wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/readme.md")
|
||
assert not wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/package.json")
|
||
assert not wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/main.go")
|
||
assert not wb.is_config_url("https://pastebin.com/raw/abc123")
|
||
assert not wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/config.yml")
|
||
assert not wb.is_config_url("https://raw.githubusercontent.com/user/repo/main/postcss.config.js")
|
||
|
||
print(" test_is_config_url: PASS")
|
||
|
||
|
||
def test_is_junk():
|
||
"""Proof: junk filter catches sample/placeholder values."""
|
||
junk_values = [
|
||
"sk-your-test-key-xxxxx",
|
||
"sk-EXAMPLEkey12345678901234567890",
|
||
"sk-your-test-api-key-CHANGEME",
|
||
"ghp_testKey1234567890abcdef1234567",
|
||
"sk-n0tr3al-1234567890123456789012345",
|
||
"AAAAAA", # < 6 unique chars
|
||
]
|
||
for v in junk_values:
|
||
assert wb.is_junk(v), f"Expected junk: {v}"
|
||
|
||
clean_values = [
|
||
"sk-proj-real-key-1234567890123456789012345678901234567890",
|
||
"ghp_realToken1234567890123456789012345678",
|
||
"sk-ant-api03-real-test-key-12345678901234567890",
|
||
"hf_zvWqKpLmRtYnBgHjUcXdEsFkAoPqRwMzXcVbNmLk",
|
||
"sk-or-v1-aBcDeFgHiJkLmNoPqRsTuVwXyZaBcDeFgHiJkLmNo",
|
||
]
|
||
for v in clean_values:
|
||
assert not wb.is_junk(v), f"Expected clean: {v}"
|
||
|
||
print(" test_is_junk: PASS")
|
||
|
||
|
||
def test_scan_and_db():
|
||
"""Proof: process_content extracts keys, writes DB, dedupes on repeat."""
|
||
# Clean the existing DB — truncate keys, reset counters
|
||
wb.conn.execute("DELETE FROM keys")
|
||
wb.total = 0
|
||
wb.seen = set()
|
||
|
||
fixture = (
|
||
"# Database Configuration\n"
|
||
"DB_HOST=localhost\n"
|
||
"DB_PORT=5432\n"
|
||
"# API Keys — real-looking values to be extracted\n"
|
||
"OPENAI_API_KEY=sk-proj-XyZ1aB2cD3eF4gH5iJ6kL7mN8oP9qR0sTuVwXyZaBcDeFgHiJkLmNoPqRsTuVw\n"
|
||
"ANTHROPIC_API_KEY=sk-ant-api03-ABCdefGHIjklMNOpqrSTUvwxYZ12345678\n"
|
||
"GITHUB_TOKEN=ghp_ABCdefGHIjklMNOpqrSTUvwxYZ1234567890\n"
|
||
"HUGGINGFACE_TOKEN=hf_ABCdefGHIjklMNOpqrSTUvwxYZ12345678\n"
|
||
"# Junk values that MUST be ignored\n"
|
||
"TEST_KEY=sk-your-test-key-xxxxx\n"
|
||
"EXAMPLE_VAR=sk-EXAMPLEkey12345678901234567890\n"
|
||
"SHORT_KEY=ab\n"
|
||
)
|
||
|
||
n = wb.process_content(fixture.encode("utf-8"),
|
||
"https://example.com/.env")
|
||
# 4 env vars × 2 extractions each (ENV_RE + PATTERN) = 8 calls,
|
||
# but each pair produces the same hash → 4 unique rows.
|
||
assert n == 8, f"Expected 8 db_add calls, got {n}"
|
||
print(f" process_content called db_add {n} times (junk+short filtered): PASS")
|
||
|
||
# Verify DB rows (hash dedup → 4 rows)
|
||
cur = wb.conn.execute(
|
||
"SELECT val, prov, found, repo FROM keys"
|
||
).fetchall()
|
||
assert len(cur) == 4, f"Expected 4 unique rows, got {len(cur)}"
|
||
for val, prov, found, repo in cur:
|
||
assert found == "wayback", f"found tag mismatch: {found}"
|
||
assert len(val) >= 28, f"key too short: {val[:20]}"
|
||
print(f" db_add wrote {n} rows with found='wayback': PASS")
|
||
|
||
# Dedup test: process same content again
|
||
n2 = wb.process_content(fixture.encode("utf-8"),
|
||
"https://example.com/.env")
|
||
cur2 = wb.conn.execute(
|
||
"SELECT val, prov, found, repo FROM keys"
|
||
).fetchall()
|
||
assert len(cur2) == 4, (
|
||
f"Dedup failed: rows before=4 after={len(cur2)}"
|
||
)
|
||
print(f" dedup on repeat: {len(cur2)} rows unchanged: PASS")
|
||
|
||
|
||
# ── Run ──────────────────────────────────────────────────────────────
|
||
|
||
if __name__ == "__main__":
|
||
test_is_config_url()
|
||
test_is_junk()
|
||
test_scan_and_db()
|
||
print("\nAll tests passed.") |