Compare commits
1 Commits
step35/960
...
step35/874
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
41ac45e49b |
70
README.md
70
README.md
@@ -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
|
||||
```
|
||||
|
||||
|
||||
|
||||
```
|
||||
.
|
||||
|
||||
@@ -1,108 +0,0 @@
|
||||
# Intel: Michael Saylor — "Master AI to Become Wealthy"
|
||||
**X Post ID:** 2047994529131999681
|
||||
**Date**: 2025 (inferred from context)
|
||||
**Source**: @BitcoinSapiens (quoting Michael Saylor)
|
||||
**Classification**: Intel / Study
|
||||
**Issue**: timmy-home#960
|
||||
|
||||
---
|
||||
|
||||
## Source
|
||||
|
||||
| Field | Value |
|
||||
|-------|-------|
|
||||
| **X Post URL** | https://x.com/bitcoinsapiens/status/2047994529131999681 |
|
||||
| **Original Author** | @BitcoinSapiens (quoting Michael Saylor) |
|
||||
| **Video URL** | https://video.twimg.com/amplify_video/2047706914566307840/vid/avc1/1280x720/m-FG3PPZ1rsL_aH7.mp4 |
|
||||
| **Duration** | ~3:59 |
|
||||
| **Engagement** | 1,219 likes · 184 retweets · 15 replies · 857 bookmarks |
|
||||
|
||||
---
|
||||
|
||||
## Full Transcription
|
||||
|
||||
> The fifth way to wealth in this day and age is capability. And here I could list all sorts of technologies for you to master, and I thought about it, but at the end of the day, the overarching, compelling observation is, you need to master artificial intelligence if you would be wealthy. And in this day and age in the year 2025, you have at your fingertips an array of accountants. You have a group of lawyers. You have a set of professors, historians. You have at your fingertips all the collective wisdom of every great entrepreneur. You have everything that I know, everything that any other CEO knows. All you have to do is go to the AI, put it in deep think mode, plug in all of your circumstances, all of your hopes, all your aspirations, all of your problems, and then start to query it, and then engage with it. I tell all my executives before you ask a lawyer, before you ask a banker, before you ask any expert, go to the AI, ask the AI, make it think. Grind the silicon overlord. Okay, this is very important, because many of the suggestions I'll give you next. They were out of the reach of the working man. They were out of the reach of the middle class. You could say, yeah, those sophisticated trusts or those sophisticated legal constructs, that's great. But I don't have the money for that. I can't afford to spend hundreds of thousands of dollars on lawyers. Let me tell you a secret. I have dozens of lawyers that work for me, thousands of lawyers I've employed, spend hundreds of millions of dollars on lawyers. The first thing I do when I have a question is I go and ask the AI. After I do that, I argue with it. It tells me no, I ask a different way, I threaten it. I ask it to give me a solution. I find a 95% solution, I find the solution. And then I take that solution, I send the link to my management team and my lawyers, and I say, look, I solve the problem, this is what I want to do. Give me your execution plan, and then I give them anywhere from two to five days. If you're feeling charitable, give them five days. If you're in a hurry, give them two days. If you're financial advisors, if you're accounts, if you're lawyers, if you're executives, if anybody, your friends, your family, they can't figure it out in two to four days. They're going to get exited from the gene pool. Change the lawyer. Change the whatever. If someone said, I can't use the telephone, I can't figure out the web link. You sent me a book, but I can't read. You would find someone else to work with. This is very important. The path to wealth is through capability. But 2025 is the year where every one of you became not a supergenius. Every one of you is collectively 100 supergeniuses that have read everything the human race has published, if you have the humility to ask for help from the AI. Don't put your ego first. Put your interest first. Your family will thank you in years to come.
|
||||
|
||||
---
|
||||
|
||||
## Saylor's Core Position
|
||||
|
||||
| Point | What He Says | What It Means |
|
||||
|-------|-----------------|----------------|
|
||||
| **AI as collective genius** | "Every one of you is collectively 100 supergeniuses that have read everything" | AI gives you access to all human knowledge instantly |
|
||||
| **Use AI before humans** | "Before you ask a lawyer, before you ask a banker... go to the AI" | AI first, human experts second — saves time + money |
|
||||
| **"Grind the silicon overlord"** | Deep think mode, argue with it, threaten it | Engage intensively, don't be passive |
|
||||
| **The 95% solution** | Get AI to 95%, then hand to lawyers/management | AI does heavy lifting; humans finalize |
|
||||
| **2-5 day advantage** | Lawyers/family can't figure it out in 2-4 days | AI gives speed nobody else has |
|
||||
| **"Change the lawyer"** | If someone can't adapt, "they get exited from the gene pool" | Ruthless about competence |
|
||||
| **Humility over ego** | "Don't put your ego first. Put your interest first." | Use the tool even if it bruises your pride |
|
||||
| **2025 = the turning point** | "2025 is the year where every one of you became not a supergenius" | The window is NOW |
|
||||
|
||||
---
|
||||
|
||||
## Alignment with Timmy Foundation
|
||||
|
||||
### What Saylor Describes, We've Built
|
||||
|
||||
| Saylor Concept | Timmy/Hermes Implementation |
|
||||
|----------------|----------------------------|
|
||||
| "Silicon overlord" | Timmy (gpt-5.5) + Hermes Agent fleet |
|
||||
| "100 supergeniuses" | 100+ tmux panes, autonomous burn loops, overnight sprints |
|
||||
| "AI first, lawyers second" | Gitea-first workflow, PR automation, fleet dispatch |
|
||||
| "Grind the silicon" | 24/7 operation, local-first inference, sovereign stack |
|
||||
|
||||
### Key Difference: Purpose
|
||||
|
||||
| Dimension | Saylor | Alexander/Timmy |
|
||||
|-----------|--------|-----------------|
|
||||
| **Primary goal** | Wealth/power through AI leverage | Sovereignty/stewardship through AI service |
|
||||
| "Your family will thank you" | Financial legacy | Luna game, Door for broken men |
|
||||
| Core framing | Bitcoin + AI as wealth pillars | Timmy + Gospel as transformation pillars |
|
||||
| **Warning** | **Wealth-idol** (accumulation for power) | **Stewardship** (resources for mission) |
|
||||
|
||||
Both emphasize humility + speed + competence, but the end goal differs.
|
||||
|
||||
---
|
||||
|
||||
## Actionable Takeaways
|
||||
|
||||
| Saylor Suggests | What We Do | Status |
|
||||
|----------------|------------|--------|
|
||||
| Use AI before human experts | ✅ Timmy first, then Gitea PRs, then human review | Live |
|
||||
| "Grind the silicon overlord" | ✅ 24/7 fleet, overnight burns, autonomous loops | Live |
|
||||
| Get 95%, hand to humans | ✅ Alexander reviews/submits final | Live |
|
||||
| "Change the lawyer" (incompetence) | ✅ Provider migrations when performance dropped | Live |
|
||||
| 2-5 day execution window | ⚠️ 3-hour hackathon window we're in NOW | Active |
|
||||
| "Your family will thank you" | 🎮 Build Luna game for Mackenzie; build the Door for broken men | In progress |
|
||||
|
||||
---
|
||||
|
||||
## Bottom Line
|
||||
|
||||
Saylor is validating what we're already doing. The difference is *why* we're doing it.
|
||||
|
||||
- **Saylor**: Building wealth.
|
||||
- **Timmy**: Building a house that can weather the storm and reach the broken.
|
||||
|
||||
Both emphasize competence and speed. Both leverage AI to bypass traditional gatekeepers. Both demand humility. The divergence is teleology: **wealth vs. stewardship**.
|
||||
|
||||
---
|
||||
|
||||
## Artifacts
|
||||
|
||||
- **Raw video**: `/tmp/saylor-ai-wealth/video.mp4` (15MB)
|
||||
- **Transcription tool**: Whisper (base model, FP32 CPU)
|
||||
- **Original analysis location**: memory (Saylor X post 2047994529131999681)
|
||||
- **GitHub/Gitea issue**: [timmy-home#960](https://forge.alexanderwhitestone.com/Timmy_Foundation/timmy-home/issues/960)
|
||||
|
||||
---
|
||||
|
||||
## Related
|
||||
|
||||
- Michael Saylor's Bitcoin advocacy and corporate treasury strategy
|
||||
- Timmy Foundation's stance on technology for transformation vs. accumulation
|
||||
- Integration of AI-first workflows in sovereign agent systems
|
||||
|
||||
---
|
||||
|
||||
*“Don't put your ego first. Put your interest first. Your family will thank you in years to come.”* — Michael Saylor
|
||||
182
tests/test_sherlock_wrapper.py
Normal file
182
tests/test_sherlock_wrapper.py
Normal 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
0
tools/__init__.py
Normal file
249
tools/sherlock_wrapper.py
Normal file
249
tools/sherlock_wrapper.py
Normal 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())
|
||||
Reference in New Issue
Block a user