Compare commits

..

1 Commits

Author SHA1 Message Date
Rockachopa
41ac45e49b feat: add Sherlock username recon wrapper with opt-in gate, caching, and normalized JSON
Some checks failed
Self-Healing Smoke / self-healing-smoke (pull_request) Failing after 24s
Agent PR Gate / gate (pull_request) Failing after 48s
Smoke Test / smoke (pull_request) Failing after 22s
Agent PR Gate / report (pull_request) Successful in 14s
- creates tools/sherlock_wrapper.py with run_sherlock() library + CLI
- opt-in gate: SHERLOCK_ENABLED=1 or --opt-in required
- local SQLite cache at ~/.cache/timmy/sherlock_cache.db (TTL: 7 days)
- normalized JSON output schema: found/missing/errors/metadata
- minimal smoke test suite: 13 tests covering schema, cache, TTL, opt-in
- adds README section with usage, schema, setup, and smoke-test instructions

Closes #874
2026-04-26 14:00:06 -04:00
14 changed files with 501 additions and 306 deletions

View File

@@ -112,6 +112,76 @@ pytest tests/
```
### Project Structure
## Sherlock Username Recon Wrapper
### Quick Usage
```bash
# Opt-in via env var
export SHERLOCK_ENABLED=1
# Or via explicit CLI flag
python -m tools.sherlock_wrapper --query "alice" --opt-in --json
# With site whitelist
python -m tools.sherlock_wrapper --query "alice" --opt-in --sites github twitter --json
```
### What It Does
Builds a bounded local wrapper around the Sherlock username OSINT tool that:
- **Opt-in gate** — SHERLOCK_ENABLED=1 or `--opt-in` required before any external call
- **Local-first caching** — results cached in `~/.cache/timmy/sherlock_cache.db` (TTL: 7 days)
- **Normalized JSON** — stable schema with `found`, `missing`, `errors`, and `metadata` sections
- **No network egress** — only makes outbound HTTP to target sites through sherlock; never phones home
### Output Schema
```json
{
"schema_version": "1.0",
"query": "alice",
"timestamp": "2025-04-26T14:23:00+00:00",
"found": [
{"site": "github", "url": "https://github.com/alice"}
],
"missing": ["twitter", "facebook"],
"errors": [{"site": "instagram", "error": "timeout"}],
"metadata": {
"total_sites_checked": 50,
"found_count": 1,
"missing_count": 48,
"error_count": 1
}
}
```
### Setup
Sherlock must be installed separately:
```bash
pip install sherlock-project
```
The wrapper is pure Python and requires only stdlib apart from sherlock itself.
### Why an Opt-In Gate?
Sherlock makes outbound HTTP requests to dozens of third-party sites. The opt-in gate:
1. Ensures a human operator explicitly approves this dependency
2. Makes the outbound traffic auditable in session logs
3. Prevents accidental invocation in automated pipelines
### Running the Smoke Test
```bash
# Run unit + integration tests
pytest tests/test_sherlock_wrapper.py -v
```
```
.

View File

@@ -1,22 +0,0 @@
---
# ansible/playbooks/deploy_mempalace.yml — Deploy MemPalace v3.0.0 to fleet wizards.
#
# Usage:
# ansible-playbook -i inventory/hosts.ini playbooks/deploy_mempalace.yml --limit ezra
# ansible-playbook -i inventory/hosts.ini playbooks/deploy_mempalace.yml
#
# Refs: Issue #570
- name: Deploy MemPalace v3.0.0 to wizard hosts
hosts: fleet
become: false
gather_facts: false
vars:
mempalace_hermes_home: "{{ ansible_env.HOME }}/.hermes"
mempalace_sessions_dir: "{{ mempalace_hermes_home }}/sessions"
mempalace_palace_path: "{{ ansible_env.HOME }}/.mempalace/palace"
mempalace_wing: "{{ inventory_hostname }}_home"
roles:
- role: ../roles/mempalace
vars:
mempalace_venv_path: "{{ ansible_env.HOME }}/.mempalace-venv"

View File

@@ -1,16 +0,0 @@
---
# MemPalace role defaults
mempalace_package_spec: "mempalace==3.0.0"
mempalace_hermes_home: "{{ ansible_env.HOME }}/.hermes"
mempalace_sessions_dir: "{{ mempalace_hermes_home }}/sessions"
mempalace_palace_path: "{{ ansible_env.HOME }}/.mempalace/palace"
mempalace_wing: "{{ inventory_hostname }}_home"
mempalace_wakeup_dir: "{{ mempalace_hermes_home }}/wakeups"
mempalace_wakeup_file: "{{ mempalace_wakeup_dir }}/{{ mempalace_wing }}.txt"
mempalace_venv_path: "{{ ansible_env.HOME }}/.mempalace-venv"
mempalace_config_path: "{{ mempalace_hermes_home }}/mempalace.yaml"
mempalace_mcp_config_path: "{{ mempalace_hermes_home }}/hermes-mcp-mempalace.yaml"
mempalace_session_hook_path: "{{ mempalace_hermes_home }}/session-start-mempalace.sh"
mempalace_run_mining: true
mempalace_run_search_test: true
mempalace_run_wake_up: true

View File

@@ -1,2 +0,0 @@
---
dependencies: []

View File

@@ -1,119 +0,0 @@
---
# MemPalace v3.0.0 deployment role for fleet wizards.
# Refs: Issue #570
- name: Ensure mempalace venv directory exists
ansible.builtin.file:
path: "{{ mempalace_venv_path }}"
state: directory
mode: '0750'
- name: Create mempalace virtual environment
ansible.builtin.command:
cmd: "python3 -m venv {{ mempalace_venv_path }}"
creates: "{{ mempalace_venv_path }}/bin/python"
- name: Install mempalace package
ansible.builtin.pip:
name: "{{ mempalace_package_spec }}"
virtualenv: "{{ mempalace_venv_path }}"
virtualenv_command: "{{ mempalace_venv_path }}/bin/python -m venv"
- name: Ensure Hermes home directory exists
ansible.builtin.file:
path: "{{ mempalace_hermes_home }}"
state: directory
mode: '0750'
- name: Ensure sessions directory exists
ansible.builtin.file:
path: "{{ mempalace_sessions_dir }}"
state: directory
mode: '0750'
- name: Ensure wakeup directory exists
ansible.builtin.file:
path: "{{ mempalace_wakeup_dir }}"
state: directory
mode: '0750'
- name: Ensure palace directory exists
ansible.builtin.file:
path: "{{ mempalace_palace_path }}"
state: directory
mode: '0750'
- name: Deploy mempalace.yaml configuration
ansible.builtin.template:
src: mempalace.yaml.j2
dest: "{{ mempalace_config_path }}"
mode: '0640'
- name: Deploy Hermes MCP mempalace config
ansible.builtin.template:
src: hermes-mcp-mempalace.yaml.j2
dest: "{{ mempalace_mcp_config_path }}"
mode: '0640'
- name: Deploy session-start wake-up hook
ansible.builtin.template:
src: session-start-mempalace.sh.j2
dest: "{{ mempalace_session_hook_path }}"
mode: '0750'
- name: Mine Hermes home directory
ansible.builtin.shell: |
set -euo pipefail
echo "" | {{ mempalace_venv_path }}/bin/mempalace mine {{ mempalace_hermes_home }} --config {{ mempalace_config_path }}
args:
executable: /bin/bash
when: mempalace_run_mining | bool
register: mine_home_result
changed_when: mine_home_result.rc == 0
- name: Mine session history
ansible.builtin.shell: |
set -euo pipefail
echo "" | {{ mempalace_venv_path }}/bin/mempalace mine {{ mempalace_sessions_dir }} --mode convos --config {{ mempalace_config_path }}
args:
executable: /bin/bash
when: mempalace_run_mining | bool
register: mine_sessions_result
changed_when: mine_sessions_result.rc == 0
- name: Run search test
ansible.builtin.shell: |
set -euo pipefail
{{ mempalace_venv_path }}/bin/mempalace search "common queries" --config {{ mempalace_config_path }} | head -20
args:
executable: /bin/bash
when: mempalace_run_search_test | bool
register: search_test_result
changed_when: false
- name: Generate wake-up context
ansible.builtin.shell: |
set -euo pipefail
{{ mempalace_venv_path }}/bin/mempalace wake-up --config {{ mempalace_config_path }} > {{ mempalace_wakeup_file }}
export HERMES_MEMPALACE_WAKEUP_FILE="{{ mempalace_wakeup_file }}"
printf '[MemPalace] wake-up context refreshed: %s\n' "$HERMES_MEMPALACE_WAKEUP_FILE"
args:
executable: /bin/bash
when: mempalace_run_wake_up | bool
register: wake_up_result
changed_when: wake_up_result.rc == 0
- name: Report MemPalace deployment summary
ansible.builtin.debug:
msg:
- "MemPalace deployed for {{ inventory_hostname }}"
- "Package: {{ mempalace_package_spec }}"
- "Config: {{ mempalace_config_path }}"
- "Palace: {{ mempalace_palace_path }}"
- "Wake-up: {{ mempalace_wakeup_file }}"
- "MCP config: {{ mempalace_mcp_config_path }}"
- "Session hook: {{ mempalace_session_hook_path }}"
- "Home mine: {{ 'OK' if mine_home_result.rc | default(1) == 0 else 'SKIPPED' }}"
- "Sessions mine: {{ 'OK' if mine_sessions_result.rc | default(1) == 0 else 'SKIPPED' }}"
- "Search test: {{ 'OK' if search_test_result.rc | default(1) == 0 else 'SKIPPED' }}"
- "Wake-up: {{ 'OK' if wake_up_result.rc | default(1) == 0 else 'SKIPPED' }}"

View File

@@ -1,6 +0,0 @@
mcp_servers:
mempalace:
command: "{{ mempalace_venv_path }}/bin/python"
args:
- -m
- mempalace.mcp_server

View File

@@ -1,21 +0,0 @@
wing: {{ mempalace_wing }}
palace: {{ mempalace_palace_path }}
rooms:
- name: sessions
description: Conversation history and durable agent transcripts
globs:
- "*.json"
- "*.jsonl"
- name: config
description: Hermes configuration and runtime settings
globs:
- "*.yaml"
- "*.yml"
- "*.toml"
- name: docs
description: Notes, markdown docs, and operating reports
globs:
- "*.md"
- "*.txt"
people: []
projects: []

View File

@@ -1,9 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
if command -v {{ mempalace_venv_path }}/bin/mempalace >/dev/null 2>&1; then
mkdir -p "{{ mempalace_wakeup_dir }}"
{{ mempalace_venv_path }}/bin/mempalace wake-up --config {{ mempalace_config_path }} > "{{ mempalace_wakeup_file }}"
export HERMES_MEMPALACE_WAKEUP_FILE="{{ mempalace_wakeup_file }}"
printf '[MemPalace] wake-up context refreshed: %s\n' "$HERMES_MEMPALACE_WAKEUP_FILE"
fi

View File

@@ -146,23 +146,6 @@ That bundle writes:
- `session-start-mempalace.sh`
- `issue-568-comment-template.md`
## Fleet Ansible deployment
Deploy MemPalace to Ezra (or the whole fleet) with the Ansible playbook:
```bash
ansible-playbook -i ansible/inventory/hosts.ini ansible/playbooks/deploy_mempalace.yml --limit ezra
```
This playbook:
1. Creates a dedicated venv and installs `mempalace==3.0.0`
2. Deploys `mempalace.yaml`, MCP config, and session-start hook
3. Mines the Hermes home and sessions directories
4. Runs a search smoke test
5. Generates the wake-up context file
Set `mempalace_run_mining=false` to skip mining on hosts where the corpus is already populated.
## Why this shape
- `wing: ezra_home` matches the issue's Ezra-specific integration target.

View File

@@ -1,92 +0,0 @@
from pathlib import Path
import unittest
ROOT = Path(__file__).resolve().parent.parent
ROLE_PATH = ROOT / "ansible" / "roles" / "mempalace"
PLAYBOOK_PATH = ROOT / "ansible" / "playbooks" / "deploy_mempalace.yml"
class TestMempalaceAnsibleRole(unittest.TestCase):
def test_role_directory_structure_exists(self):
self.assertTrue(ROLE_PATH.exists(), "mempalace role directory missing")
for subdir in ["tasks", "templates", "defaults", "meta"]:
self.assertTrue(
(ROLE_PATH / subdir).exists(),
f"mempalace role subdir missing: {subdir}",
)
def test_role_defaults_contains_required_variables(self):
defaults_path = ROLE_PATH / "defaults" / "main.yml"
self.assertTrue(defaults_path.exists())
text = defaults_path.read_text(encoding="utf-8")
required_vars = [
"mempalace_package_spec",
"mempalace_hermes_home",
"mempalace_sessions_dir",
"mempalace_palace_path",
"mempalace_wing",
"mempalace_wakeup_dir",
"mempalace_wakeup_file",
"mempalace_venv_path",
"mempalace_config_path",
"mempalace_mcp_config_path",
"mempalace_session_hook_path",
"mempalace_run_mining",
"mempalace_run_search_test",
"mempalace_run_wake_up",
]
for var in required_vars:
self.assertIn(var, text, f"missing default var: {var}")
def test_role_tasks_contain_required_steps(self):
tasks_path = ROLE_PATH / "tasks" / "main.yml"
self.assertTrue(tasks_path.exists())
text = tasks_path.read_text(encoding="utf-8")
required_steps = [
"Create mempalace virtual environment",
"Install mempalace package",
"Deploy mempalace.yaml configuration",
"Deploy Hermes MCP mempalace config",
"Deploy session-start wake-up hook",
"Mine Hermes home directory",
"Mine session history",
"Run search test",
"Generate wake-up context",
]
for step in required_steps:
self.assertIn(step, text, f"missing task: {step}")
def test_role_templates_are_valid(self):
yaml_template = ROLE_PATH / "templates" / "mempalace.yaml.j2"
mcp_template = ROLE_PATH / "templates" / "hermes-mcp-mempalace.yaml.j2"
hook_template = ROLE_PATH / "templates" / "session-start-mempalace.sh.j2"
self.assertTrue(yaml_template.exists())
self.assertTrue(mcp_template.exists())
self.assertTrue(hook_template.exists())
yaml_text = yaml_template.read_text(encoding="utf-8")
self.assertIn("wing: {{ mempalace_wing }}", yaml_text)
self.assertIn("palace: {{ mempalace_palace_path }}", yaml_text)
self.assertIn("rooms:", yaml_text)
mcp_text = mcp_template.read_text(encoding="utf-8")
self.assertIn("mcp_servers:", mcp_text)
self.assertIn("mempalace:", mcp_text)
self.assertIn("mempalace.mcp_server", mcp_text)
hook_text = hook_template.read_text(encoding="utf-8")
self.assertIn("mempalace wake-up", hook_text)
self.assertIn("HERMES_MEMPALACE_WAKEUP_FILE", hook_text)
def test_playbook_exists_and_targets_fleet(self):
self.assertTrue(PLAYBOOK_PATH.exists(), "deploy_mempalace.yml playbook missing")
text = PLAYBOOK_PATH.read_text(encoding="utf-8")
self.assertIn("hosts: fleet", text)
self.assertIn("../roles/mempalace", text)
self.assertIn("mempalace_venv_path", text)
if __name__ == "__main__":
unittest.main()

View File

@@ -85,8 +85,6 @@ class TestMempalaceEzraIntegration(unittest.TestCase):
"mcp_servers:",
"HERMES_MEMPALACE_WAKEUP_FILE",
"Metrics reply for #568",
"Fleet Ansible deployment",
"ansible-playbook",
]
for snippet in required:
self.assertIn(snippet, text)

View File

@@ -0,0 +1,182 @@
#!/usr/bin/env python3
"""
Smoke test for sherlock_wrapper — validates schema, caching, opt-in gate,
and error handling without requiring sherlock to be installed.
"""
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch, MagicMock
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "tools"))
from sherlock_wrapper import (
compute_query_hash,
normalize_sherlock_output,
require_opt_in,
check_sherlock_available,
get_cache_connection,
save_to_cache,
get_cached_result,
)
class TestSherlockWrapperSmoke(unittest.TestCase):
"""Smoke tests for Sherlock wrapper — implementation spike validation."""
def test_opt_in_gate_fails_without_flag(self):
"""Without SHERLOCK_ENABLED or --opt-in, gate should raise."""
with patch("sherlock_wrapper.SHERLOCK_ENABLED", False):
with self.assertRaises(RuntimeError) as ctx:
require_opt_in(opt_in=False)
self.assertIn("opt-in only", str(ctx.exception).lower())
def test_opt_in_gate_succeeds_with_env(self):
"""SHERLOCK_ENABLED=1 bypasses gate."""
with patch("sherlock_wrapper.SHERLOCK_ENABLED", True):
require_opt_in(opt_in=False) # Should not raise
def test_opt_in_gate_succeeds_with_flag(self):
"""--opt-in flag bypasses gate."""
with patch("sherlock_wrapper.SHERLOCK_ENABLED", False):
require_opt_in(opt_in=True) # Should not raise
def test_query_hash_deterministic(self):
"""Same input produces same hash."""
h1 = compute_query_hash("alice")
h2 = compute_query_hash("alice")
self.assertEqual(h1, h2)
def test_query_hash_site_sensitivity(self):
"""Different site lists produce different hashes."""
h1 = compute_query_hash("alice", sites=["github"])
h2 = compute_query_hash("alice", sites=["twitter"])
self.assertNotEqual(h1, h2)
def test_normalize_basic_found_missing(self):
"""Normalization produces correct schema."""
raw = {
"github": {"status": "found", "url": "https://github.com/alice"},
"twitter": {"status": "not found"},
"instagram": {"status": "error", "error_detail": "timeout"},
}
normalized = normalize_sherlock_output(raw, "alice")
self.assertEqual(normalized["query"], "alice")
self.assertEqual(normalized["metadata"]["found_count"], 1)
self.assertEqual(normalized["metadata"]["missing_count"], 1)
self.assertEqual(normalized["metadata"]["error_count"], 1)
self.assertEqual(len(normalized["found"]), 1)
self.assertEqual(normalized["found"][0]["site"], "github")
self.assertIn("twitter", normalized["missing"])
self.assertEqual(normalized["errors"][0]["site"], "instagram")
def test_normalized_schema_has_required_fields(self):
"""Output schema contains all required top-level keys."""
raw = {"site1": {"status": "not found"}}
normalized = normalize_sherlock_output(raw, "testuser")
required = ["schema_version", "query", "timestamp", "found", "missing",
"errors", "metadata"]
for key in required:
self.assertIn(key, normalized)
self.assertIsInstance(normalized["timestamp"], str)
self.assertIsInstance(normalized["found"], list)
self.assertIsInstance(normalized["missing"], list)
self.assertIsInstance(normalized["errors"], list)
self.assertIsInstance(normalized["metadata"], dict)
def test_cache_roundtrip(self):
"""Result can be written and read back from cache."""
with tempfile.TemporaryDirectory() as tmp:
with patch("sherlock_wrapper.CACHE_DB", Path(tmp) / "cache.db"):
test_result = {
"schema_version": "1.0",
"query": "alice",
"timestamp": "2025-04-26T00:00:00+00:00",
"found": [],
"missing": ["github"],
"errors": [],
"metadata": {"total_sites_checked": 1, "found_count": 0, "missing_count": 1, "error_count": 0},
}
query_hash = compute_query_hash("alice")
save_to_cache(query_hash, test_result)
retrieved = get_cached_result(query_hash)
self.assertEqual(retrieved, test_result)
def test_cache_miss_on_stale(self):
"""Cache returns None when entry is older than 7 days."""
with tempfile.TemporaryDirectory() as tmp:
db_path = Path(tmp) / "cache.db"
with patch("sherlock_wrapper.CACHE_DB", db_path):
old_ts = "2025-04-01T00:00:00+00:00"
old_result = {
"schema_version": "1.0", "query": "alice",
"timestamp": old_ts, "found": [], "missing": [], "errors": [],
"metadata": {"total_sites_checked": 0, "found_count": 0, "missing_count": 0, "error_count": 0},
}
query_hash = compute_query_hash("alice")
# Direct DB insert with controlled timestamp (bypass save_to_cache's NOW)
conn = get_cache_connection()
conn.execute(
"INSERT INTO cache (query_hash, result_json, timestamp) VALUES (?, ?, ?)",
(query_hash, json.dumps(old_result), old_ts)
)
conn.commit()
retrieved = get_cached_result(query_hash)
self.assertIsNone(retrieved)
def test_sherlock_available_check(self):
"""check_sherlock_available returns bool."""
available = check_sherlock_available()
self.assertIsInstance(available, bool)
# Note: on this test system sherlock may not be installed, so False is expected.
# The important thing is the function returns a bool.
print(f"[INFO] Sherlock installed: {available}")
class TestSherlockWrapperIntegration(unittest.TestCase):
"""Integration tests with mocked sherlock module."""
def test_run_sherlock_with_opt_in(self):
"""run_sherlock succeeds with opt-in and returns normalized result."""
fake_sherlock = MagicMock()
fake_sherlock.sherlock = MagicMock(return_value={
"github": {"status": "found", "url": "https://github.com/alice"},
"twitter": {"status": "not found"},
})
with patch.dict("sys.modules", {"sherlock": fake_sherlock}):
import importlib
import sherlock_wrapper
importlib.reload(sherlock_wrapper)
with patch.dict(os.environ, {"SHERLOCK_ENABLED": "1"}):
from sherlock_wrapper import run_sherlock
result = run_sherlock("alice", opt_in=True)
self.assertEqual(result["query"], "alice")
self.assertEqual(result["metadata"]["found_count"], 1)
def test_run_sherlock_fails_without_opt_in(self):
"""run_sherlock raises RuntimeError without opt-in."""
from sherlock_wrapper import run_sherlock
with self.assertRaises(RuntimeError) as ctx:
run_sherlock("alice", opt_in=False)
self.assertIn("opt-in only", str(ctx.exception).lower())
def test_run_sherlock_uses_cache(self):
"""Cached result short-circuits sherlock execution."""
cached = {
"schema_version": "1.0", "query": "alice", "timestamp": "2025-04-26T00:00:00+00:00",
"found": [{"site": "github", "url": "https://github.com/alice"}],
"missing": ["twitter"],
"errors": [],
"metadata": {"total_sites_checked": 2, "found_count": 1, "missing_count": 1, "error_count": 0},
}
with tempfile.TemporaryDirectory() as tmp:
with patch("sherlock_wrapper.CACHE_DB", Path(tmp) / "cache.db"):
query_hash = compute_query_hash("alice")
save_to_cache(query_hash, cached)
from sherlock_wrapper import run_sherlock
result = run_sherlock("alice", opt_in=True)
self.assertEqual(result, cached)

0
tools/__init__.py Normal file
View File

249
tools/sherlock_wrapper.py Normal file
View File

@@ -0,0 +1,249 @@
#!/usr/bin/env python3
"""
Sherlock username recon wrapper — opt-in, cached, normalized JSON output.
This is an implementation spike (issue #874) to validate local integration
of the Sherlock OSINT tool without violating sovereignty/provenance standards.
"""
import argparse
import hashlib
import json
import os
import sqlite3
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional, Dict, Any, List
# Opt-in gate: must have SHERLOCK_ENABLED=1 or --opt-in flag
SHERLOCK_ENABLED = os.environ.get("SHERLOCK_ENABLED", "0") == "1"
# Cache location
CACHE_DIR = Path.home() / ".cache" / "timmy"
CACHE_DB = CACHE_DIR / "sherlock_cache.db"
# Normalized output schema version
SCHEMA_VERSION = "1.0"
def require_opt_in(opt_in: bool = False) -> None:
"""Enforce opt-in gate for Sherlock external dependency."""
if not (SHERLOCK_ENABLED or opt_in):
raise RuntimeError(
"Sherlock is opt-in only. Set SHERLOCK_ENABLED=1 or pass --opt-in."
)
def check_sherlock_available() -> bool:
"""Check if sherlock Python package is installed."""
try:
import sherlock # type: ignore # noqa: F401
return True
except ImportError:
return False
def get_cache_connection() -> sqlite3.Connection:
"""Initialize cache directory and return DB connection."""
CACHE_DIR.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(CACHE_DB))
conn.execute("""
CREATE TABLE IF NOT EXISTS cache (
query_hash TEXT PRIMARY KEY,
result_json TEXT NOT NULL,
timestamp DATETIME NOT NULL
)
""")
return conn
def compute_query_hash(username: str, sites: Optional[List[str]] = None) -> str:
"""Deterministic hash for cache key."""
components = [username.lower().strip()]
if sites:
components.extend(sorted(sites))
raw = "|".join(components)
return hashlib.sha256(raw.encode()).hexdigest()
def get_cached_result(query_hash: str) -> Optional[Dict[str, Any]]:
"""Retrieve cached result if available and not stale (TTL: 7 days)."""
conn = get_cache_connection()
cur = conn.execute(
"SELECT result_json, timestamp FROM cache WHERE query_hash = ?",
(query_hash,)
)
row = cur.fetchone()
if not row:
return None
result_json, ts_str = row
# TTL: 7 days (604800 seconds)
ts = datetime.fromisoformat(ts_str)
age_seconds = (datetime.now(timezone.utc) - ts).total_seconds()
if age_seconds >= 604800:
return None
return json.loads(result_json)
def save_to_cache(query_hash: str, result: Dict[str, Any]) -> None:
"""Persist result to cache."""
conn = get_cache_connection()
conn.execute(
"INSERT OR REPLACE INTO cache (query_hash, result_json, timestamp) VALUES (?, ?, ?)",
(query_hash, json.dumps(result), datetime.now(timezone.utc).isoformat())
)
conn.commit()
conn.close()
def normalize_sherlock_output(
raw_result: Dict[str, Any],
username: str,
sites_checked: Optional[List[str]] = None
) -> Dict[str, Any]:
"""
Convert raw sherlock output into a stable, normalized schema.
Expected sherlock result shape (via Python API):
{
"site_name": {"url": "...", "status": "found"|"not found"|"error", ...},
...
}
"""
found: List[Dict[str, str]] = []
missing: List[str] = []
errors: List[Dict[str, str]] = []
for site_name, site_data in raw_result.items():
status = site_data.get("status", "")
url = site_data.get("url", "")
if status == "found" and url:
found.append({"site": site_name, "url": url})
elif status == "not found":
missing.append(site_name)
else:
errors.append({"site": site_name, "error": status or "unknown"})
# Compute totals from the original site list if provided
total_sites = len(raw_result) if sites_checked is None else len(sites_checked)
return {
"schema_version": SCHEMA_VERSION,
"query": username,
"timestamp": datetime.now(timezone.utc).isoformat(),
"found": found,
"missing": missing,
"errors": errors,
"metadata": {
"total_sites_checked": total_sites,
"found_count": len(found),
"missing_count": len(missing),
"error_count": len(errors),
},
}
def run_sherlock(
username: str,
sites: Optional[List[str]] = None,
timeout: Optional[int] = None,
opt_in: bool = False
) -> Dict[str, Any]:
"""
Execute Sherlock wrapper with opt-in gate, caching, and normalization.
"""
require_opt_in(opt_in)
# Compute cache key
query_hash = compute_query_hash(username, sites)
# Check cache first — avoids dependency requirement on cache hit
cached = get_cached_result(query_hash)
if cached is not None:
return cached
# Only require sherlock on cache miss
if not check_sherlock_available():
raise RuntimeError(
"Sherlock Python package not installed. "
"Install with: pip install sherlock-project"
)
# Call sherlock
try:
import sherlock
from sherlock import sherlock as sherlock_main # type: ignore
if sites:
result = sherlock_main(username, site_list=sites, timeout=timeout or 10)
else:
result = sherlock_main(username, timeout=timeout or 10)
normalized = normalize_sherlock_output(result, username, sites)
save_to_cache(query_hash, normalized)
return normalized
except Exception as e:
raise RuntimeError(f"Sherlock execution failed: {e}") from e
def main() -> int:
parser = argparse.ArgumentParser(
description="Sherlock username OSINT wrapper — opt-in, cached, normalized JSON"
)
parser.add_argument(
"--query", "-q", required=True,
help="Username to search across sites"
)
parser.add_argument(
"--opt-in", action="store_true",
help="Explicit opt-in flag (alternatively set SHERLOCK_ENABLED=1)"
)
parser.add_argument(
"--sites", "-s", nargs="+",
help="Specific sites to check (default: all supported)"
)
parser.add_argument(
"--timeout", "-t", type=int, default=10,
help="Request timeout per site (default: 10)"
)
parser.add_argument(
"--json", action="store_true",
help="Output normalized JSON to stdout"
)
parser.add_argument(
"--no-cache",
action="store_true",
help="Bypass cached result (if any)"
)
args = parser.parse_args()
try:
result = run_sherlock(
username=args.query,
sites=args.sites,
timeout=args.timeout,
opt_in=args.opt_in
)
if args.json:
print(json.dumps(result, indent=2))
else:
print(f"Query: {result['query']}")
print(f"Found: {result['metadata']['found_count']} site(s)")
print(f"Missing: {result['metadata']['missing_count']} site(s)")
print(f"Errors: {result['metadata']['error_count']} site(s)")
for f in result['found']:
print(f" [{f['site']}] {f['url']}")
return 0
except RuntimeError as e:
print(f"ERROR: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())