diff --git a/eagleosint/audit.py b/eagleosint/audit.py new file mode 100644 index 0000000..99f954d --- /dev/null +++ b/eagleosint/audit.py @@ -0,0 +1,64 @@ +"""Append-only audit log for all provider queries. + +Every provider execution is recorded as a JSON Lines entry. +The log is immutable by design: opened in append-only mode. +Queries are SHA256-hashed to avoid storing PII in plaintext. +""" +from __future__ import annotations + +import hashlib +import json +import os +import uuid +from datetime import datetime, timezone +from typing import Any + +from platformdirs import user_data_dir + +# Audit log lives in the data directory, not config +_DATA_DIR = user_data_dir("eagleosint", appauthor=False) +AUDIT_LOG_PATH = os.path.join(_DATA_DIR, "audit.jsonl") + +os.makedirs(_DATA_DIR, exist_ok=True) + +# Session ID: unique per process run - groups all queries from one session +SESSION_ID = uuid.uuid4().hex[:12] + +def _hash_query(query: str) -> str: + """SHA256 hash of the query string - PII never stored in plaintext.""" + return hashlib.sha256(query.encode("utf-8")).hexdigest() + +def audit_log( + event: str, + provider: str, + query: str, + *, + result_count: int | None = None, + success: bool = True, + extra: dict[str, Any] | None = None +) -> None: + """Append a single audit entry to the log file. + + Args: + event: Event type -- "query_start", "query_end", "query_error". + provider: Provider name (e.g. "github", "userrecon"). + query: Raw query string — will be hashed before writing. + result_count: Number of results returned (for query_end). + success: Whether the query succeeded. + extra: Optional additional metadata. + """ + entry: dict[str, Any] = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "session_id": SESSION_ID, + "event": event, + "provider": provider, + "query_hash": _hash_query(query), + "success": success, + } + if result_count is not None: + entry["result_count"] = result_count + if extra is not None: + entry["extra"] = extra + + with open(AUDIT_LOG_PATH, "a", encoding="utf-8") as f: + f.write(json.dumps(entry, separators=(",", ":")) + "\n") \ No newline at end of file diff --git a/eagleosint/plugin.py b/eagleosint/plugin.py index d9d2cfd..1e7be63 100644 --- a/eagleosint/plugin.py +++ b/eagleosint/plugin.py @@ -7,15 +7,18 @@ UserReconProvider, GoDorkerProvider. Pending: Facebook (deprecated API — isolate/remove), TempMail (polling pattern). """ - from __future__ import annotations +import logging from abc import ABC, abstractmethod from enum import Enum -from typing import ClassVar +from typing import Any, ClassVar from eagleosint.models import ProviderResult +logger = logging.getLogger(__name__) + + class ProviderCategory(str, Enum): USERNAME = "username" EMAIL = "email" @@ -36,9 +39,10 @@ class BaseProvider(ABC): Contract every provider plugin must satisfy. Class-level attributes declare metadata. - execute() is the single required async entry point. + execute() is the abstract entry point that subclasses implement. + run() wraps execute() with automatic audit logging. - Rules of implementors: + Rules for implementors: - execute() must return a list (empty on failure, never raise) - execute() must catch all internal exceptions and log them - required_keys must list every config key the provider needs @@ -55,30 +59,53 @@ class BaseProvider(ABC): # ---------------------------------------------------------- @abstractmethod - async def execute(self, query: str) -> list[ProviderResult]: + def execute(self, query: str, **kwargs: Any) -> list[ProviderResult]: """ Run the provider against a query string. Return a list of ProviderResult subclass instances. - Return an empty list on any failure - never raise. + Return an empty list on any failure — never raise. + """ + + # ---------------------------------------------------------- + # Audited entry point + # ---------------------------------------------------------- + + def run(self, query: str, **kwargs: Any) -> list[ProviderResult]: + """Execute the provider with automatic audit logging. + + This is the recommended way to call a provider. It wraps + execute() with query_start/query_end/query_error audit events. """ + from eagleosint.audit import audit_log + + audit_log("query_start", self.name, query) + try: + results = self.execute(query, **kwargs) + audit_log( + "query_end", self.name, query, + result_count=len(results), success=True, + ) + return results + except Exception as exc: + logger.error("provider %s failed: %s", self.name, exc) + audit_log( + "query_error", self.name, query, + success=False, extra={"error": str(exc)}, + ) + return [] # ---------------------------------------------------------- # Concrete helpers (override if needed) # ---------------------------------------------------------- def is_available(self) -> bool: - """ - True if all required API keys are present in CONFIGS. - Override for providers whose availability depends on other factors. - """ + """True if all required API keys are present in config.""" from eagleosint.config import settings return all(settings.get_key(k) for k in self.required_keys) def validate_query(self, query: str) -> str | None: - """ - Optional pre-execution validation. - Return None if the query is acceptable. - Return an error message string if it should be rejected. + """Optional pre-execution validation. + Return None if acceptable, error message string if not. """ if not query or not query.strip(): return "Query must not be empty" diff --git a/eagleosint/providers/bitly.py b/eagleosint/providers/bitly.py index 80e562d..a89654b 100644 --- a/eagleosint/providers/bitly.py +++ b/eagleosint/providers/bitly.py @@ -66,7 +66,7 @@ def bypass_bitly() -> URLExpansion | None: ).strip() provider = BitlyProvider() - results = provider.execute(bitly_url) + results = provider.run(bitly_url) if not results: print(f"{RED}No results for '{bitly_url}'.{WHITE}") @@ -74,7 +74,7 @@ def bypass_bitly() -> URLExpansion | None: getpass(SPACE_PREFIX + "press enter for back to previous menu ") return None - result = results[0] + result: URLExpansion = results[0] # type: ignore[assignment] print( f"{SPACE_PREFIX}{BG_BLUE} DONE {WHITE} Original URL: " f"\u001b[38;5;32m{result.original_url}" diff --git a/eagleosint/providers/github.py b/eagleosint/providers/github.py index d56e4d2..1702ecc 100644 --- a/eagleosint/providers/github.py +++ b/eagleosint/providers/github.py @@ -76,7 +76,7 @@ def github_lookup() -> GitHubProfile | None: print(WHITE) provider = GithubProvider() - results = provider.execute(github_user) + results = provider.run(github_user) if not results: print(f"{RED}No results for '{github_user}'.{WHITE}") @@ -84,7 +84,7 @@ def github_lookup() -> GitHubProfile | None: getpass(SPACE_PREFIX + "press enter for back to previous menu ") return None - profile = results[0] + profile: GitHubProfile = results[0] # type: ignore[assignment] table_data = [[str(key), str(value)] for key, value in profile.raw.items()] for line_item in tabulate( diff --git a/eagleosint/providers/godorker.py b/eagleosint/providers/godorker.py index 695d52b..aa1f71a 100644 --- a/eagleosint/providers/godorker.py +++ b/eagleosint/providers/godorker.py @@ -98,7 +98,7 @@ def _on_result(result: DorkResult) -> None: try: provider = GoDorkerProvider() - results = provider.execute(dork_query, on_result=_on_result) + results: list[DorkResult] = provider.run(dork_query, on_result=_on_result) # type: ignore[assignment] except KeyboardInterrupt: print(f"{RED}Dorking aborted by user.{WHITE}") results = [] diff --git a/eagleosint/providers/mailfinder.py b/eagleosint/providers/mailfinder.py index 63d153f..d8b635a 100644 --- a/eagleosint/providers/mailfinder.py +++ b/eagleosint/providers/mailfinder.py @@ -184,7 +184,7 @@ def _on_result(result: EmailResult, current: int, total: int) -> None: try: provider = MailFinderProvider() - results = provider.execute( + results: list[EmailResult] = provider.run( # type: ignore[assignment] full_name, api_key=api_key, on_result=_on_result ) except KeyboardInterrupt: diff --git a/eagleosint/providers/network.py b/eagleosint/providers/network.py index 1060f4d..7ac1532 100644 --- a/eagleosint/providers/network.py +++ b/eagleosint/providers/network.py @@ -127,7 +127,7 @@ def iplocation() -> IPResult | None: ip_address = input(f"{SPACE_PREFIX}{BLUE}>{WHITE} enter IP:{BLUE} ").strip() provider = IPLocationProvider() - results = provider.execute(ip_address) + results = provider.run(ip_address) print(WHITE + LINES_SEPARATOR) if not results: @@ -136,7 +136,7 @@ def iplocation() -> IPResult | None: getpass(SPACE_PREFIX + "press enter for back to previous menu ") return None - result = results[0] + result: IPResult = results[0] # type: ignore[assignment] for label, value in [ ("IP", result.ip), ("CITY", result.city), ("COUNTRY", result.country), ("LOC", result.coordinates), ("ORG", result.org), ("TIMEZONE", result.timezone), @@ -154,7 +154,7 @@ def infoga(option: str) -> DomainResult | None: return None provider = DomainInfoProvider() - results = provider.execute(target, query_type=option) + results = provider.run(target, query_type=option) print(WHITE + LINES_SEPARATOR) if not results: @@ -163,9 +163,10 @@ def infoga(option: str) -> DomainResult | None: getpass(SPACE_PREFIX + "press enter for back to previous menu ") return None - for line in results[0].records: + result: DomainResult = results[0] # type: ignore[assignment] + for line in result.records: print(f"{SPACE_PREFIX}{BLUE}-{WHITE} {line}") print(WHITE + LINES_SEPARATOR) getpass(SPACE_PREFIX + "press enter for back to previous menu ") - return results[0] \ No newline at end of file + return result \ No newline at end of file diff --git a/eagleosint/providers/phoneinfo.py b/eagleosint/providers/phoneinfo.py index 43cf778..98578e4 100644 --- a/eagleosint/providers/phoneinfo.py +++ b/eagleosint/providers/phoneinfo.py @@ -81,7 +81,7 @@ def phoneinfo() -> PhoneResult | None: save_config() provider = PhoneInfoProvider() - results = provider.execute(phone_number, api_key=api_key) + results = provider.run(phone_number, api_key=api_key) print(WHITE + LINES_SEPARATOR) if not results: @@ -90,7 +90,7 @@ def phoneinfo() -> PhoneResult | None: getpass(SPACE_PREFIX + "press enter for back to previous menu ") return None - result = results[0] + result: PhoneResult = results[0] # type: ignore[assignment] if result.raw: for info_key, info_value in result.raw.items(): print( diff --git a/eagleosint/providers/userrecon.py b/eagleosint/providers/userrecon.py index a9d5d0b..1581c1d 100644 --- a/eagleosint/providers/userrecon.py +++ b/eagleosint/providers/userrecon.py @@ -161,7 +161,7 @@ def _on_result(hit: AccountHit, current: int, total: int) -> None: print(WHITE + LINES_SEPARATOR) provider = UserReconProvider() - results = provider.execute(username, on_result=_on_result) + results: list[AccountHit] = provider.run(username, on_result=_on_result) # type: ignore[assignment] print() print(WHITE + LINES_SEPARATOR) diff --git a/tests/test_audit.py b/tests/test_audit.py new file mode 100644 index 0000000..18c73e8 --- /dev/null +++ b/tests/test_audit.py @@ -0,0 +1,109 @@ +"""Tests for eagleosint.audit module.""" +from __future__ import annotations + +import json +import os + +import pytest + +from eagleosint.audit import audit_log, _hash_query, AUDIT_LOG_PATH, SESSION_ID + + +@pytest.fixture() +def tmp_audit_log(tmp_path, monkeypatch): + """Redirect audit log to a temporary file.""" + log_path = str(tmp_path / "audit.jsonl") + monkeypatch.setattr("eagleosint.audit.AUDIT_LOG_PATH", log_path) + return log_path + + +class TestHashQuery: + + def test_deterministic(self): + assert _hash_query("test") == _hash_query("test") + + def test_different_inputs_differ(self): + assert _hash_query("alice") != _hash_query("bob") + + def test_returns_hex_string(self): + h = _hash_query("test") + assert len(h) == 64 + assert all(c in "0123456789abcdef" for c in h) + + +class TestAuditLog: + + def test_writes_jsonl_entry(self, tmp_audit_log): + audit_log("query_start", "github", "octocat") + with open(tmp_audit_log, "r") as f: + lines = f.readlines() + assert len(lines) == 1 + entry = json.loads(lines[0]) + assert entry["event"] == "query_start" + assert entry["provider"] == "github" + assert entry["session_id"] == SESSION_ID + assert entry["success"] is True + assert "timestamp" in entry + + def test_query_is_hashed_not_plaintext(self, tmp_audit_log): + audit_log("query_start", "github", "secret_username") + with open(tmp_audit_log, "r") as f: + content = f.read() + assert "secret_username" not in content + assert _hash_query("secret_username") in content + + def test_append_mode(self, tmp_audit_log): + audit_log("query_start", "github", "user1") + audit_log("query_end", "github", "user1", result_count=3) + with open(tmp_audit_log, "r") as f: + lines = f.readlines() + assert len(lines) == 2 + assert json.loads(lines[0])["event"] == "query_start" + assert json.loads(lines[1])["event"] == "query_end" + assert json.loads(lines[1])["result_count"] == 3 + + def test_error_event(self, tmp_audit_log): + audit_log( + "query_error", "github", "user1", + success=False, extra={"error": "timeout"}, + ) + with open(tmp_audit_log, "r") as f: + entry = json.loads(f.readline()) + assert entry["success"] is False + assert entry["extra"]["error"] == "timeout" + + def test_session_id_consistent(self, tmp_audit_log): + audit_log("query_start", "github", "a") + audit_log("query_start", "bitly", "b") + with open(tmp_audit_log, "r") as f: + lines = f.readlines() + id1 = json.loads(lines[0])["session_id"] + id2 = json.loads(lines[1])["session_id"] + assert id1 == id2 + + +class TestProviderRunAudit: + """Test that BaseProvider.run() triggers audit logging.""" + + def test_run_logs_start_and_end(self, tmp_audit_log, monkeypatch): + from eagleosint.providers.github import GithubProvider + from unittest.mock import MagicMock + + mock_resp = MagicMock() + mock_resp.json.return_value = {"login": "test", "name": "Test"} + mock_resp.raise_for_status.return_value = None + monkeypatch.setattr( + "eagleosint.providers.github._session.get", + lambda *a, **kw: mock_resp, + ) + + provider = GithubProvider() + results = provider.run("testuser") + + assert len(results) == 1 + with open(tmp_audit_log, "r") as f: + lines = f.readlines() + assert len(lines) == 2 + assert json.loads(lines[0])["event"] == "query_start" + assert json.loads(lines[1])["event"] == "query_end" + assert json.loads(lines[1])["result_count"] == 1 \ No newline at end of file