refactor(security): harden api_key_leak_scanner (redaction, streaming, CI exit codes)

python_scripts/api_key_leak_scanner.py

@@ -1,54 +1,194 @@
 #!/usr/bin/env python3
+"""Scan a directory tree for hardcoded secrets and credentials.
+
+The scanner walks a workspace, matches the contents of every text file against
+a set of well-known credential signatures (generic API keys, Google API keys,
+AWS access keys, Slack tokens and GitHub personal access tokens), and reports
+each hit. It is intended to run locally and as a lightweight CI gate before
+code is committed or pushed.
+
+Exit codes:
+    0 - scan completed and no potential secrets were found
+    1 - scan completed and at least one potential secret was found
+    2 - the scan could not run (for example, the target path does not exist)
+
+Safety note:
+    Matched secret material is never echoed to the console. Only a redacted
+    context snippet is printed, so scanner output can be pasted into CI logs or
+    bug reports without re-leaking the very credential it detected.
+"""
+
+import argparse
 import os
 import re
 import sys
+from pathlib import Path
+
+# --- Configuration ----------------------------------------------------------
+
+# Directories that never hold first-party source and are expensive or pointless
+# to walk (VCS metadata, virtual environments, caches, vendored dependencies).
+SKIP_DIRS = {
+    ".git", ".hg", ".svn", ".tox", ".venv", "venv", "__pycache__", "node_modules",
+}
+
+# Binary and archive extensions that must not be read as text.
+SKIP_EXTENSIONS = {
+    ".png", ".jpg", ".jpeg", ".gif", ".ico", ".bmp", ".tiff",
+    ".exe", ".dll", ".so", ".dylib", ".bin", ".o", ".a",
+    ".pyc", ".pyo", ".class", ".jar",
+    ".zip", ".gz", ".tgz", ".bz2", ".xz", ".tar", ".7z", ".rar", ".pdf",
+}
+
+# Files larger than this are skipped: they are almost never source code and
+# reading them would waste memory. Keeps the scanner's footprint bounded.
+MAX_FILE_BYTES = 5 * 1024 * 1024  # 5 MiB
+
+# Named credential signatures. The *last* capturing group of each pattern is
+# treated as the secret value so it can be redacted before anything is printed.
+SECRET_PATTERNS = {
+    "Generic API Key": re.compile(
+        r"(?i)(api[_-]?key|apikey|auth[_-]?token|access[_-]?token)"
+        r"[\s:=]+['\"]([a-zA-Z0-9]{16,})['\"]"
+    ),
+    "Google API Key": re.compile(r"(AIza[0-9A-Za-z\-_]{35})"),
+    "AWS Access Key": re.compile(r"(AKIA[0-9A-Z]{16})"),
+    "Slack Token": re.compile(r"(xox[baprs]-[0-9a-zA-Z]{10,48})"),
+    "GitHub Personal Access Token": re.compile(r"(ghp_[a-zA-Z0-9]{36})"),
+}
+
+REDACTION_MARKER = "***REDACTED***"
+
+
+def _secret_span(match):
+    """Return the ``(start, end)`` offsets of the secret inside ``match``.
+
+    The secret is the last non-empty capturing group, or the whole match when
+    the pattern defines no capturing groups.
+    """
+    for index in range(match.re.groups, 0, -1):
+        if match.group(index) is not None:
+            return match.start(index), match.end(index)
+    return match.start(0), match.end(0)
+
+
+def mask_line(line, match):
+    """Return ``line`` with the matched secret replaced by a fixed marker.
+
+    The surrounding context is preserved so a reviewer can locate the leak, but
+    the credential itself is never part of the returned string.
+    """
+    start, end = _secret_span(match)
+    redacted = f"{line[:start]}{REDACTION_MARKER}{line[end:]}"
+    return redacted.strip()
+
+
+def _looks_binary(path):
+    """Return True when the first block of ``path`` contains a NUL byte."""
+    try:
+        with path.open("rb") as probe:
+            return b"\x00" in probe.read(1024)
+    except OSError:
+        # If we cannot even read the header, let the text pass below report it.
+        return False
 
-def scan_for_secrets(directory):
+
+def scan_file(path):
+    """Scan a single file and return a list of finding dictionaries.
+
+    Each finding is ``{"name", "path", "line", "snippet"}`` where ``snippet``
+    is redacted. Binary files, oversized files and unreadable files return an
+    empty list instead of raising, so one bad file cannot abort a whole scan.
+
+    Args:
+        path: A :class:`pathlib.Path` to a regular file.
+
+    Returns:
+        A list of finding dictionaries (possibly empty).
     """
-    Regex-driven string analyzer that scans code workspaces to flag 
-    leaked hardcoded authorization variables.
+    findings = []
+    path = Path(path)
+    try:
+        if path.suffix.lower() in SKIP_EXTENSIONS:
+            return findings
+
+        size = path.stat().st_size
+        if size > MAX_FILE_BYTES:
+            print(f"[!] Skipping large file ({size} bytes): {path}")
+            return findings
+
+        if _looks_binary(path):
+            return findings
+
+        with path.open("r", encoding="utf-8", errors="ignore") as handle:
+            for line_no, line in enumerate(handle, start=1):
+                for name, pattern in SECRET_PATTERNS.items():
+                    for match in pattern.finditer(line):
+                        findings.append({
+                            "name": name,
+                            "path": str(path),
+                            "line": line_no,
+                            "snippet": mask_line(line, match),
+                        })
+    except OSError as exc:
+        print(f"[!] Could not read {path}: {exc}")
+    return findings
+
+
+def scan_directory(directory):
+    """Recursively scan ``directory`` and return every finding.
+
+    Skipped directories are pruned in place so :func:`os.walk` never descends
+    into them.
+
+    Args:
+        directory: Root directory to walk (str or :class:`pathlib.Path`).
+
+    Returns:
+        A flat list of finding dictionaries across all scanned files.
     """
-    print(f"=== API Key Leak Scanner ===")
-    print(f"[*] Scanning directory: {directory}")
-
-    # Common patterns for secrets
-    patterns = {
-        "Generic API Key": r"(?i)(api[_-]key|apikey|auth[_-]token|access[_-]token)[\s:=]+['\"]([a-zA-Z0-9]{16,})['\"]",
-        "Google API Key": r"AIza[0-9A-Za-z\\-_]{35}",
-        "AWS Access Key": r"AKIA[0-9A-Z]{16}",
-        "Slack Token": r"xox[baprs]-[0-9a-zA-Z]{10,48}",
-        "GitHub Personal Access Token": r"ghp_[a-zA-Z0-9]{36}"
-    }
-
-    leaks_found = 0
-    for root, dirs, files in os.walk(directory):
-        # Skip hidden directories like .git
-        dirs[:] = [d for d in dirs if not d.startswith('.')]
-
-        for file in files:
-            file_path = os.path.join(root, file)
-
-            # Skip binary files
-            if file.endswith(('.png', '.jpg', '.exe', '.pyc', '.zip')):
-                continue
-
-            try:
-                with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
-                    content = f.read()
-                    for name, pattern in patterns.items():
-                        matches = re.finditer(pattern, content)
-                        for match in matches:
-                            print(f"[!] LEAK DETECTED [{name}] in {file_path}")
-                            print(f"    Match: {match.group(0)[:30]}...")
-                            leaks_found += 1
-            except Exception as e:
-                print(f"[!] Could not read {file_path}: {e}")
-
-    if leaks_found == 0:
+    root = Path(directory)
+    findings = []
+    for current, dirs, files in os.walk(root):
+        dirs[:] = sorted(d for d in dirs if d not in SKIP_DIRS)
+        for name in sorted(files):
+            findings.extend(scan_file(Path(current) / name))
+    return findings
+
+
+def report(findings):
+    """Print a human-readable report for ``findings``."""
+    for finding in findings:
+        print(f"[!] LEAK DETECTED [{finding['name']}] "
+              f"{finding['path']}:{finding['line']}")
+        print(f"    {finding['snippet']}")
+    if not findings:
         print("[+] No hardcoded API keys or secrets detected.")
     else:
-        print(f"[*] Scan complete. {leaks_found} potential leaks found.")
+        print(f"[*] Scan complete. {len(findings)} potential leaks found.")
+
+
+def main(argv=None):
+    """Parse arguments, run the scan and return a process exit code."""
+    parser = argparse.ArgumentParser(
+        description="Scan a directory tree for hardcoded API keys and secrets.")
+    parser.add_argument(
+        "directory", nargs="?", default=".",
+        help="directory to scan (default: current directory)")
+    args = parser.parse_args(argv)
+
+    target = Path(args.directory)
+    print("=== API Key Leak Scanner ===")
+    print(f"[*] Scanning directory: {target}")
+
+    if not target.is_dir():
+        print(f"[-] Error: {target} is not a directory.")
+        return 2
+
+    findings = scan_directory(target)
+    report(findings)
+    return 1 if findings else 0
+
 
 if __name__ == "__main__":
-    target_dir = sys.argv[1] if len(sys.argv) > 1 else "."
-    scan_for_secrets(target_dir)
+    sys.exit(main())

tests/conftest.py

@@ -0,0 +1,13 @@
+"""Test configuration for the kali-linux-automation script suite.
+
+The scripts live in a flat ``python_scripts/`` directory rather than an
+installable package, so the directory is added to ``sys.path`` to make them
+importable from the tests.
+"""
+
+import sys
+from pathlib import Path
+
+PYTHON_SCRIPTS = Path(__file__).resolve().parent.parent / "python_scripts"
+if str(PYTHON_SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(PYTHON_SCRIPTS))

tests/test_api_key_leak_scanner.py

@@ -0,0 +1,141 @@
+"""Tests for python_scripts/api_key_leak_scanner.py.
+
+Focus areas:
+    * detection of the supported credential signatures
+    * redaction - a matched secret must never appear in the output
+    * failure handling - bad paths, binary files and oversized files
+    * the exit-code contract that makes the scanner usable as a CI gate
+"""
+
+import api_key_leak_scanner as scanner
+
+# Well-known, non-functional example credentials (these are not real secrets).
+AWS_KEY = "AKIAIOSFODNN7EXAMPLE"
+GITHUB_TOKEN = "ghp_" + "A" * 36
+GOOGLE_KEY = "AIza" + "B" * 35
+SLACK_TOKEN = "xoxb-" + "1" * 12
+
+
+def _write(path, text):
+    path.write_text(text, encoding="utf-8")
+    return path
+
+
+# --- Detection --------------------------------------------------------------
+
+def test_scan_file_detects_aws_key(tmp_path):
+    target = _write(tmp_path / "config.py", f'aws_access_key = "{AWS_KEY}"\n')
+    findings = scanner.scan_file(target)
+    assert len(findings) == 1
+    assert findings[0]["name"] == "AWS Access Key"
+    assert findings[0]["line"] == 1
+
+
+def test_scan_file_detects_generic_api_key(tmp_path):
+    target = _write(tmp_path / "settings.txt", 'api_key = "abcdef1234567890XYZ"\n')
+    findings = scanner.scan_file(target)
+    assert [f["name"] for f in findings] == ["Generic API Key"]
+
+
+def test_scan_file_detects_github_token(tmp_path):
+    target = _write(tmp_path / "ci.sh", f'export TOKEN={GITHUB_TOKEN}\n')
+    names = {f["name"] for f in scanner.scan_file(target)}
+    assert "GitHub Personal Access Token" in names
+
+
+def test_scan_file_detects_google_and_slack_keys(tmp_path):
+    target = _write(tmp_path / "keys.env",
+                    f"GOOGLE={GOOGLE_KEY}\nSLACK={SLACK_TOKEN}\n")
+    names = {f["name"] for f in scanner.scan_file(target)}
+    assert {"Google API Key", "Slack Token"} <= names
+
+
+def test_scan_directory_finds_secrets_in_nested_files(tmp_path):
+    nested = tmp_path / "pkg" / "deep"
+    nested.mkdir(parents=True)
+    _write(nested / "creds.py", f'key = "{AWS_KEY}"\n')
+    findings = scanner.scan_directory(tmp_path)
+    assert any(f["name"] == "AWS Access Key" for f in findings)
+
+
+def test_clean_directory_has_no_findings(tmp_path):
+    _write(tmp_path / "ok.py", "print('nothing to see here')\n")
+    assert scanner.scan_directory(tmp_path) == []
+
+
+# --- Redaction --------------------------------------------------------------
+
+def test_scan_file_redacts_the_secret(tmp_path):
+    target = _write(tmp_path / "config.py", f'aws_access_key = "{AWS_KEY}"\n')
+    snippet = scanner.scan_file(target)[0]["snippet"]
+    assert AWS_KEY not in snippet
+    assert scanner.REDACTION_MARKER in snippet
+
+
+def test_mask_line_removes_only_the_secret():
+    pattern = scanner.SECRET_PATTERNS["AWS Access Key"]
+    line = f'access_key = "{AWS_KEY}"  # do not commit\n'
+    match = pattern.search(line)
+    masked = scanner.mask_line(line, match)
+    assert AWS_KEY not in masked
+    assert "do not commit" in masked
+
+
+# --- Skip rules and failure handling ---------------------------------------
+
+def test_binary_file_is_skipped(tmp_path):
+    target = tmp_path / "blob.py"
+    target.write_bytes(b"\x00\x01\x02" + AWS_KEY.encode())
+    assert scanner.scan_file(target) == []
+
+
+def test_oversized_file_is_skipped(tmp_path, monkeypatch):
+    target = _write(tmp_path / "big.txt", "x")
+    monkeypatch.setattr(scanner, "MAX_FILE_BYTES", 0)
+    assert scanner.scan_file(target) == []
+
+
+def test_skipped_extension_is_ignored(tmp_path):
+    target = tmp_path / "archive.zip"
+    target.write_bytes(AWS_KEY.encode())
+    assert scanner.scan_file(target) == []
+
+
+def test_git_directory_is_skipped(tmp_path):
+    git_dir = tmp_path / ".git"
+    git_dir.mkdir()
+    _write(git_dir / "config", f'key = "{AWS_KEY}"\n')
+    assert scanner.scan_directory(tmp_path) == []
+
+
+def test_hidden_env_file_is_still_scanned(tmp_path):
+    _write(tmp_path / ".env", f'API_KEY="{AWS_KEY}"\n')
+    findings = scanner.scan_directory(tmp_path)
+    assert any(f["name"] == "AWS Access Key" for f in findings)
+
+
+# --- Exit-code contract -----------------------------------------------------
+
+def test_main_returns_one_when_a_secret_is_found(tmp_path, capsys):
+    _write(tmp_path / "a.py", f'key = "{AWS_KEY}"\n')
+    assert scanner.main([str(tmp_path)]) == 1
+    out = capsys.readouterr().out
+    assert "LEAK DETECTED" in out
+    assert AWS_KEY not in out
+
+
+def test_main_returns_zero_for_a_clean_tree(tmp_path):
+    _write(tmp_path / "a.py", "print('clean')\n")
+    assert scanner.main([str(tmp_path)]) == 0
+
+
+def test_main_returns_two_for_a_missing_directory(tmp_path, capsys):
+    missing = tmp_path / "does-not-exist"
+    assert scanner.main([str(missing)]) == 2
+    assert "is not a directory" in capsys.readouterr().out
+
+
+def test_main_defaults_to_current_directory(tmp_path, monkeypatch, capsys):
+    _write(tmp_path / "a.py", f'key = "{AWS_KEY}"\n')
+    monkeypatch.chdir(tmp_path)
+    assert scanner.main([]) == 1