Compare commits

..

1 Commits

Author SHA1 Message Date
Alexander Whitestone
aa69610a9b [P1] Sonnet workforce — full end-to-end smoke test (#512)
Some checks failed
Self-Healing Smoke / self-healing-smoke (pull_request) Failing after 26s
Smoke Test / smoke (pull_request) Failing after 30s
Agent PR Gate / gate (pull_request) Failing after 49s
Agent PR Gate / report (pull_request) Successful in 9s
**New scripts:**
- scripts/sonnet-smoke-test.sh — Validates Sonnet can clone via Gitea HTTP,
  branch, commit, push, create PR, and verify PR state via API.
  Run: ./scripts/sonnet-smoke-test.sh [--cleanup]
  All 6 checks pass (clone, branch, commit, push, PR create, PR verify).

- scripts/agent-dispatch.sh — One-shot prompt generator for fleet workers.
  Supports: sonnet, claude, kimi, grok, gemini, ezra, bezalel, allegro, timmy.
  Usage: ./scripts/agent-dispatch.sh <agent> <repo> <issue#> [<org>]

**Uni-Wizard v4:**
- Added SONNET to House enum (uni-wizard/v4/uni_wizard/__init__.py)

**Proof:**
Smoke test executed successfully, creating and verifying PR #856
(#856),
which was then closed to keep the repo clean.

Closes #512
2026-04-22 02:29:38 -04:00
7 changed files with 307 additions and 501 deletions

View File

@@ -112,76 +112,6 @@ 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
```
```
.

111
scripts/agent-dispatch.sh Executable file
View File

@@ -0,0 +1,111 @@
#!/bin/bash
# ============================================================================
# Agent Dispatch — One-shot prompt generator for fleet workers
# ============================================================================
# Refs: timmy-home #512
#
# Packages context, token, repo, issue, and Git/Gitea commands into a
# copy-pasteable prompt for any agent (Claude, Sonnet, Kimi, Grok, etc.).
#
# Usage:
# scripts/agent-dispatch.sh <agent> <repo> <issue#> [<org>]
#
# Supported agents:
# sonnet, claude, kimi, grok, gemini, ezra, bezalel, allegro, timmy
#
# Example:
# scripts/agent-dispatch.sh sonnet the-nexus 844 Timmy_Foundation
# ============================================================================
set -euo pipefail
AGENT="${1:-}"
REPO="${2:-}"
ISSUE="${3:-}"
ORG="${4:-Timmy_Foundation}"
TOKEN="${GITEA_TOKEN:-$(cat ~/.config/gitea/token 2>/dev/null || true)}"
FORGE="https://forge.alexanderwhitestone.com"
if [ -z "$AGENT" ] || [ -z "$REPO" ] || [ -z "$ISSUE" ]; then
echo "Usage: $0 <agent> <repo> <issue#> [<org>]"
echo ""
echo "Supported agents:"
echo " sonnet — Anthropic Claude Sonnet (cloud, high-reasoning)"
echo " claude — Anthropic Claude (general)"
echo " kimi — Moonshot Kimi K2.5 (cloud, long-context)"
echo " grok — xAI Grok (cloud, real-time)"
echo " gemini — Google Gemini (cloud, multimodal)"
echo " ezra — Local archivist house (read-before-write)"
echo " bezalel — Local artificer house (proof-required)"
echo " allegro — Local dispatch house (tempo-and-routing)"
echo " timmy — Local sovereign house (final review)"
exit 1
fi
# Validate agent
VALID_AGENTS="sonnet claude kimi grok gemini ezra bezalel allegro timmy"
if ! echo "$VALID_AGENTS" | grep -qw "$AGENT"; then
echo "ERROR: Unknown agent '$AGENT'"
echo "Valid agents: $VALID_AGENTS"
exit 1
fi
# Fetch issue details
if [ -n "$TOKEN" ]; then
ISSUE_JSON=$(curl -s -H "Authorization: token ${TOKEN}" \
"${FORGE}/api/v1/repos/${ORG}/${REPO}/issues/${ISSUE}" 2>/dev/null || true)
ISSUE_TITLE=$(echo "$ISSUE_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get('title',''))" 2>/dev/null || true)
ISSUE_BODY=$(echo "$ISSUE_JSON" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get('body',''))" 2>/dev/null || true)
else
echo "WARNING: No Gitea token found. Issue details will be blank."
ISSUE_TITLE=""
ISSUE_BODY=""
fi
cat <<EOF
================================================================================
DISPATCH PROMPT — ${AGENT} → ${ORG}/${REPO}#${ISSUE}
================================================================================
Agent: ${AGENT}
Repo: ${ORG}/${REPO}
Issue: #${ISSUE}
Title: ${ISSUE_TITLE}
--- ISSUE BODY ---
${ISSUE_BODY}
--- INSTRUCTIONS ---
1. Clone the repo:
git clone --depth 1 "https://\${TOKEN}@forge.alexanderwhitestone.com/${ORG}/${REPO}.git"
cd ${REPO}
2. Create branch:
git checkout -b ${AGENT}/${REPO}-${ISSUE}
3. Read the issue, implement the fix or feature.
4. Test your changes locally.
5. Commit and push:
git add -A
git commit -m "[${AGENT}] ${ISSUE_TITLE} (#${ISSUE})"
git push origin ${AGENT}/${REPO}-${ISSUE}
6. Open PR via Gitea API:
curl -X POST \\
-H "Authorization: token \${TOKEN}" \\
-H "Content-Type: application/json" \\
"${FORGE}/api/v1/repos/${ORG}/${REPO}/pulls" \\
-d '{"title":"[${AGENT}] ${ISSUE_TITLE}","head":"${AGENT}/${REPO}-${ISSUE}","base":"main","body":"Closes #${ISSUE}"}'
7. File new issues for anything discovered.
Token: \${GITEA_TOKEN} or ~/.config/gitea/token
Forge: ${FORGE}
Sovereignty and service always.
================================================================================
EOF

195
scripts/sonnet-smoke-test.sh Executable file
View File

@@ -0,0 +1,195 @@
#!/bin/bash
# ============================================================================
# Sonnet Workforce Smoke Test
# ============================================================================
# Refs: timmy-home #512
#
# Validates that the Sonnet workforce agent can perform the full
# clone → code → commit → push → PR workflow via Gitea HTTP.
#
# Usage:
# scripts/sonnet-smoke-test.sh [--cleanup]
#
# Exit codes:
# 0 — all checks passed
# 1 — one or more checks failed
# ============================================================================
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
TOKEN="${GITEA_TOKEN:-$(cat ~/.config/gitea/token 2>/dev/null || true)}"
FORGE="https://forge.alexanderwhitestone.com"
ORG="Timmy_Foundation"
REPO="timmy-home"
TEST_BRANCH="smoke/sonnet-$(date +%s)"
# Colors
GREEN='\\033[0;32m'
RED='\\033[0;31m'
YELLOW='\\033[0;33m'
NC='\\033[0m'
PASS=0
FAIL=0
log_pass() { echo -e "${GREEN}${NC} $1"; PASS=$((PASS + 1)); }
log_fail() { echo -e "${RED}${NC} $1"; FAIL=$((FAIL + 1)); }
log_info() { echo -e "${YELLOW}${NC} $1"; }
# ── Prerequisites ──────────────────────────────────────────────────────────────────────────────────────
log_info "Checking prerequisites..."
if [ -z "$TOKEN" ]; then
log_fail "Gitea token not found (checked GITEA_TOKEN env and ~/.config/gitea/token)"
exit 1
fi
if ! command -v git &>/dev/null; then
log_fail "git not installed"
exit 1
fi
if ! command -v curl &>/dev/null; then
log_fail "curl not installed"
exit 1
fi
if ! command -v python3 &>/dev/null; then
log_fail "python3 not installed"
exit 1
fi
log_pass "Prerequisites OK"
# ── 1. Clone via Gitea HTTP ───────────────────────────────────────────────────────────────────────────────────────────────────────
log_info "Step 1: Clone repo via Gitea HTTP..."
TMPDIR=$(mktemp -d)
CLONE_URL="${FORGE}/${ORG}/${REPO}.git"
cd "$TMPDIR"
if git clone --depth 1 "https://${TOKEN}@${FORGE#https://}/${ORG}/${REPO}.git" smoke-clone 2>/dev/null; then
log_pass "Clone via Gitea HTTP"
else
log_fail "Clone via Gitea HTTP"
rm -rf "$TMPDIR"
exit 1
fi
# ── 2. Commit ─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
log_info "Step 2: Create branch and commit..."
cd "$TMPDIR/smoke-clone"
git checkout -b "$TEST_BRANCH" 2>/dev/null || true
# Make a harmless change
printf "# Sonnet smoke test marker\\n# timestamp: %s\\n" "$(date -u +%Y-%m-%dT%H:%M:%SZ)" > SONNET_SMOKE_MARKER.md
git add SONNET_SMOKE_MARKER.md
if git -c user.email="sonnet@timmy.local" -c user.name="Sonnet Smoke Test" \
commit -m "test: sonnet smoke test marker" 2>/dev/null; then
log_pass "Commit created"
else
log_fail "Commit failed"
rm -rf "$TMPDIR"
exit 1
fi
# ── 3. Push ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
log_info "Step 3: Push branch..."
if git push origin "$TEST_BRANCH" 2>/dev/null; then
log_pass "Push to origin"
else
log_fail "Push to origin"
rm -rf "$TMPDIR"
exit 1
fi
# ── 4. Create PR ───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
log_info "Step 4: Create PR via Gitea API..."
PR_RESPONSE=$(curl -s -X POST \
-H "Authorization: token ${TOKEN}" \
-H "Content-Type: application/json" \
"${FORGE}/api/v1/repos/${ORG}/${REPO}/pulls" \
-d "{
\"title\": \"test: sonnet smoke test ${TEST_BRANCH}\",
\"head\": \"${TEST_BRANCH}\",
\"base\": \"main\",
\"body\": \"Automated smoke test verifying Sonnet can clone, commit, push, and open a PR.\\n\\nRefs #512\"
}" 2>/dev/null)
PR_NUMBER=$(echo "$PR_RESPONSE" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get('number',''))")
if [ -n "$PR_NUMBER" ] && [ "$PR_NUMBER" != "None" ]; then
log_pass "PR created (#${PR_NUMBER})"
PR_URL="${FORGE}/${ORG}/${REPO}/pulls/${PR_NUMBER}"
echo " URL: $PR_URL"
else
log_fail "PR creation failed"
echo " Response: $PR_RESPONSE"
rm -rf "$TMPDIR"
exit 1
fi
# ── 5. Verify PR exists ──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
log_info "Step 5: Verify PR exists via API..."
PR_CHECK=$(curl -s -H "Authorization: token ${TOKEN}" \
"${FORGE}/api/v1/repos/${ORG}/${REPO}/pulls/${PR_NUMBER}" 2>/dev/null)
PR_STATE=$(echo "$PR_CHECK" | python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get('state',''))")
if [ "$PR_STATE" = "open" ]; then
log_pass "PR verified open via API"
else
log_fail "PR state is '$PR_STATE', expected 'open'"
fi
# ── Cleanup (optional) ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
if [ "${1:-}" = "--cleanup" ]; then
log_info "Cleaning up smoke test artifacts..."
curl -s -X PATCH -H "Authorization: token ${TOKEN}" \
-H "Content-Type: application/json" \
"${FORGE}/api/v1/repos/${ORG}/${REPO}/pulls/${PR_NUMBER}" \
-d '{"state":"closed"}' >/dev/null 2>&1 || true
git push origin --delete "$TEST_BRANCH" 2>/dev/null || true
log_pass "Cleanup complete"
fi
rm -rf "$TMPDIR"
# ── Summary ────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────
echo ""
echo "================================================================"
echo " Sonnet Smoke Test Summary"
echo "================================================================"
echo -e " Passed: ${GREEN}${PASS}${NC}"
echo -e " Failed: ${RED}${FAIL}${NC}"
echo ""
if [ "$FAIL" -gt 0 ]; then
echo -e "${RED}RESULT: FAILED${NC}"
exit 1
else
echo -e "${GREEN}RESULT: PASSED${NC}"
echo ""
echo "Sonnet workforce is verified end-to-end:"
echo " ✓ Clone via Gitea HTTP"
echo " ✓ Branch + commit"
echo " ✓ Push to origin"
echo " ✓ Open PR via API"
echo " ✓ Verify PR state"
exit 0
fi

View File

@@ -1,182 +0,0 @@
#!/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)

View File

View File

@@ -1,249 +0,0 @@
#!/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())

View File

@@ -38,6 +38,7 @@ class House(Enum):
EZRA = "ezra" # Archivist, reader
BEZALEL = "bezalel" # Artificer, builder
ALLEGRO = "allegro" # Tempo-and-dispatch, connected
SONNET = "sonnet" # Anthropic Claude Sonnet (cloud, high-reasoning)
class Mode(Enum):