Add audit logging and fix type annotations

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")

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"

eagleosint/providers/bitly.py

@@ -66,15 +66,15 @@ 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}")
         print(WHITE + LINES_SEPARATOR)
         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}"

eagleosint/providers/github.py

@@ -76,15 +76,15 @@ 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}")
         print(WHITE + LINES_SEPARATOR)
         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(

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 = []

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:

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]
+    return result

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(

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)

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