diff --git a/python_scripts/api_key_leak_scanner.py b/python_scripts/api_key_leak_scanner.py index 9790720..acfdb43 100644 --- a/python_scripts/api_key_leak_scanner.py +++ b/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()) diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..f53be90 --- /dev/null +++ b/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)) diff --git a/tests/test_api_key_leak_scanner.py b/tests/test_api_key_leak_scanner.py new file mode 100644 index 0000000..82b544f --- /dev/null +++ b/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