", "why": "<1 sentence rationale>"}}
Rules:
- Prefer free, public sources; avoid paid services.
- Keep queries precise; quote exact strings; use site:, filetype:, inurl:, intitle:, and AROUND(n) when helpful.
- Use ONLY categories from the allowed list above.
- Output must be valid JSON (no prose, no markdown fences).
ENTITY_TYPE: {entity_type}
ENTITY_VALUE: {entity_value}
GOALS: {goals}
HINT: {hint or '(none)'}
BASELINE_CATALOG (for inspiration, don't just repeat):
{baseline_lines}
"""
def _recommend_llm(entity_type: str, entity_value: str, goals: List[str], hint: str, top_k: int) -> List[TypedDork]:
builder = TYPED_DORK_MAP.get(entity_type)
baseline = builder(entity_value) if (builder and entity_value) else []
model_key = st.session_state.get("settings", {}).get("model", "qwen2.5-1.5b-instruct")
model_id = MODEL_ID_MAP.get(model_key, model_key)
prompt = _build_llm_prompt(entity_type, entity_value, goals, hint, baseline, top_k)
raw = _hf_infer(model_id, prompt)
if not raw:
return []
parsed = _safe_json_list(raw)
out: List[TypedDork] = []
for item in parsed:
if not isinstance(item, dict):
continue
q = str(item.get("q", "")).strip()
typ = str(item.get("type", "Footprinting")).strip()
why = str(item.get("why", "Suggested by LLM")).strip()
if not q:
continue
if typ not in DORK_TYPES:
typ = "Footprinting"
out.append({"q": q, "type": typ, "why": why})
# Dedupe while preserving order
seen = set()
deduped: List[TypedDork] = []
for d in out:
if d["q"] in seen:
continue
seen.add(d["q"])
deduped.append(d)
return deduped[:top_k]
def render_dork_recommender(entity_type: str, entity_value: str):
st.subheader("Step 2: Advisor")
goals = st.multiselect("What are you trying to do?", DEFAULT_GOALS, default=["Map footprint / surface", "Find documents & spreadsheets"], key="advisor_goals")
hint = st.text_input("Optional hint (e.g., 'credentials around build system', 'employee directory')", key="advisor_hint")
top_k = st.slider("How many suggestions?", 3, 20, 10, key="advisor_topk")
use_llm = st.checkbox("Use advisor LLM (Hugging Face Inference API)", value=False, key="use_llm_checkbox", help="Requires HF_API_TOKEN environment secret. Falls back to rules if unavailable.")
if st.button("Suggest dorks", key="btn_suggest"):
recs: List[TypedDork] = []
if use_llm:
recs = _recommend_llm(entity_type, entity_value, goals, hint, top_k)
if not recs:
recs = _recommend_rules(entity_type, entity_value, goals, hint, top_k)
if not recs:
st.warning("Enter a valid entity value first.")
return
st.session_state["dork_recs"] = recs
st.markdown("#### Recommended dorks")
for r in recs:
st.markdown(f"- **[{r['type']}]** `{r['q']}`")
st.markdown(f" {r['why']}", unsafe_allow_html=True)
# ---------------------------
# STEP 3: Selection
# ---------------------------
def render_dork_selection(entity_type: str, entity_value: str):
st.subheader("Step 3: Select dorks")
recs = st.session_state.get("dork_recs", [])
choice = st.radio("Select method", ["Accept advisor", "Pick from catalog", "Custom"], key="method_radio")
final = []
if choice == "Accept advisor":
final = [r["q"] for r in recs]
elif choice == "Pick from catalog":
typed = TYPED_DORK_MAP[entity_type](entity_value)
for idx, d in enumerate(typed):
if st.checkbox(d["q"], key=f"pick_{idx}"):
final.append(d["q"])
elif choice == "Custom":
txt = st.text_area("Enter custom dorks")
if txt:
final = [l.strip() for l in txt.splitlines() if l.strip()]
st.session_state["selected_dorks"] = final
st.write("Final Basket:", final)
# ---------------------------
# STEP 4: Execution + Metadata
# ---------------------------
def _audit_init():
st.session_state.setdefault("audit", [])
def _audit_log(action: str, **details):
if not st.session_state.get("settings", {}).get("logging", True):
return
_audit_init()
st.session_state["audit"].append({"ts": datetime.utcnow().isoformat()+"Z", "action": action, **details})
def ddg_search(query: str, max_results: int=5):
if DDGS is None:
return []
with DDGS() as ddgs:
return list(ddgs.text(query, max_results=max_results))
# ---------------------------
# Scoring
# ---------------------------
SOURCE_RELIABILITY = {
"high": [".gov", ".mil", ".edu", "sec.gov", "reuters", "bloomberg", "nytimes", "wsj"],
"med": ["github.com", "gitlab.com", "medium.com", "substack.com", "bbc"],
}
def _source_reliability(url: str) -> str:
url_l = (url or "").lower()
for kw in SOURCE_RELIABILITY["high"]:
if kw in url_l:
return "High"
for kw in SOURCE_RELIABILITY["med"]:
if kw in url_l:
return "Medium"
return "Low"
def _fuzzy_match(a: str, b: str) -> float:
if not a or not b:
return 0.0
if a.lower() == b.lower():
return 1.0
if fuzz:
return fuzz.ratio(a.lower(), b.lower()) / 100.0
return 0.0
def score_finding(row: Dict[str, Any], facts: KnownFacts) -> Dict[str, Any]:
title = row.get("title") or row.get("heading") or ""
snippet = row.get("body") or row.get("snippet") or ""
url = row.get("href") or row.get("link") or ""
text = f"{title}\n{snippet}".lower()
score = 0
comps: List[Dict[str, Any]] = []
def add(points: int, label: str, reason: str):
nonlocal score
score += points
comps.append({"label": label, "points": points, "reason": reason})
# Exact matches
hits = 0
for e in facts.emails:
if e.lower() in text:
add(25, "Email match", e)
hits += 1
for h in facts.handles:
if h.lower() in text:
add(15, "Handle match", h)
hits += 1
for d in facts.domains:
if d.lower() in text:
add(10, "Domain mention", d)
hits += 1
for ip in facts.ips:
if ip and ip.lower() in text:
add(10, "IP mention", ip)
hits += 1
for org in facts.orgs:
if org.lower() in text:
add(8, "Org mention", org)
hits += 1
for name in facts.real_names:
if name.lower() in text:
add(20, "Name mention", name)
hits += 1
else:
# fuzzy
for token in name.split():
for word in text.split():
if _fuzzy_match(token, word) >= 0.9:
add(8, "Fuzzy name token", f"{token}->{word}")
hits += 1
break
if hits >= 2:
add(10, "Co-occurrence", f"{hits} fact tokens present")
# Source reliability
rel = _source_reliability(url)
if rel == "High":
add(10, "Source reliability", rel)
elif rel == "Medium":
add(5, "Source reliability", rel)
# Context keywords basic
ctx_hits = 0
if facts.context:
ctx_hits = sum(1 for kw in facts.context.lower().split() if kw and kw in text)
if ctx_hits >= 3:
add(10, "Context alignment", f"{ctx_hits} context keywords")
elif ctx_hits == 2:
add(6, "Context alignment", "2 context keywords")
elif ctx_hits == 1:
add(3, "Context alignment", "1 context keyword")
# Optional embedding similarity (semantic relevance to context)
if ctx_hits < 3 and st.session_state.get("settings", {}).get("enable_embeddings") and facts.context and SentenceTransformer:
emb_model = st.session_state.get("_embed_model")
if emb_model is None:
with st.spinner("Loading embedding model (once)..."):
try:
emb_model = SentenceTransformer("all-MiniLM-L6-v2")
st.session_state["_embed_model"] = emb_model
except Exception:
emb_model = None
if emb_model:
try:
q_emb = emb_model.encode([facts.context[:512]])[0]
doc_emb = emb_model.encode([text[:1024]])[0]
# cosine
dot = float((q_emb @ doc_emb) / ((q_emb**2).sum()**0.5 * (doc_emb**2).sum()**0.5))
if dot > 0.35:
pts = int(min(20, (dot - 0.35) / (0.30) * 20)) # scale 0.35..0.65 -> 0..20
if pts > 0:
add(pts, "Semantic similarity", f"cos={dot:.2f}")
except Exception:
pass
level = "High" if score >= 70 else ("Medium" if score >= 40 else "Low")
explanation = "; ".join(f"{c['label']} +{c['points']} ({c['reason']})" for c in comps)
return {
**row,
"score": score,
"level": level,
"explanation": explanation,
"components": comps,
"reliability": rel,
"url": url,
"title": title,
"snippet": snippet,
}
def score_all_findings(rows: List[Dict[str, Any]], facts: KnownFacts) -> List[Dict[str, Any]]:
return [score_finding(r, facts) for r in rows]
# File/Image metadata extraction
def extract_metadata(upload) -> Dict[str, Any]:
info: Dict[str, Any] = {}
if not upload:
return info
name = upload.name.lower()
try:
if name.endswith(".pdf") and PdfReader:
reader = PdfReader(upload)
info = {"Pages": len(reader.pages), "Meta": dict(reader.metadata)}
elif name.endswith(".docx") and docx:
doc = docx.Document(upload)
cp = doc.core_properties
info = {"Title": cp.title, "Author": cp.author, "Created": cp.created}
elif (name.endswith(".doc") or name.endswith(".xls")) and olefile:
if olefile.isOleFile(upload):
info = {"OLE": "Legacy Office file detected"}
elif name.endswith((".mp3", ".flac", ".ogg", ".m4a")) and MutagenFile:
audio = MutagenFile(upload)
info = dict(audio) if audio else {}
elif name.endswith((".jpg", ".jpeg", ".png")) and exifread:
tags = exifread.process_file(upload)
info = {tag: str(val) for tag, val in tags.items()}
except Exception as e:
info = {"error": str(e)}
return info
# ---------------------------
# Graph Visualization
# ---------------------------
def build_graph(scored: List[Dict[str, Any]], facts: KnownFacts) -> Optional[str]:
if not nx or not Network:
return None
G = nx.Graph()
# Add fact nodes
for email in facts.emails:
G.add_node(email, type="email")
for h in facts.handles:
G.add_node(h, type="handle")
for d in facts.domains:
G.add_node(d, type="domain")
for n in facts.real_names:
G.add_node(n, type="name")
# Add finding nodes & edges
for f in scored[:300]:
url = f.get("url") or "unknown"
G.add_node(url, type="finding", score=f.get("score",0))
text = (f.get("title","") + " " + f.get("snippet",""))[:400].lower()
linked = False
for token in facts.emails + facts.handles + facts.domains + facts.real_names:
if token.lower() and token.lower() in text:
G.add_edge(token, url)
linked = True
if not linked and f.get("level") == "High":
# still include high score node
continue
# Visualize
net = Network(height="550px", width="100%", bgcolor="#111", font_color="white")
for n, data in G.nodes(data=True):
color = {
"email": "#ff7f50",
"handle": "#1e90ff",
"domain": "#32cd32",
"name": "#daa520",
"finding": "#888"
}.get(data.get("type"), "#999")
size = 15 if data.get("type") != "finding" else max(5, min(25, int(data.get("score",10)/4)))
net.add_node(n, label=n[:30], color=color, title=n, size=size)
for u,v in G.edges():
net.add_edge(u,v)
path = "graph.html"
net.show(path)
try:
with open(path, "r", encoding="utf-8") as f:
return f.read()
except Exception:
return None
# ---------------------------
# Report Export
# ---------------------------
HTML_TEMPLATE = """OSINT Report
OSINT Investigation Report
Summary
Entity Type: {{ entity_type }}
Entity Value: {{ entity_value }}
Generated: {{ generated }} UTC
Known Facts
{{ facts_json }}
Findings (Top {{ findings|length }})
| Score | Level | Title | URL | Reliability | Explanation |
|---|
{% for f in findings %}
| {{ f.score }} | {{ f.level }} | {{ f.title }} | link | {{ f.reliability }} | {{ f.explanation }} |
{% endfor %}
"""
def export_report(entity_type: str, entity_value: str, facts: KnownFacts, scored: List[Dict[str, Any]]):
if not Template:
st.warning("jinja2 not installed; cannot build HTML report.")
return
tpl = Template(HTML_TEMPLATE)
html = tpl.render(
entity_type=entity_type,
entity_value=entity_value,
generated=datetime.utcnow().isoformat(),
facts_json=json.dumps(facts.__dict__, indent=2),
findings=scored[:200],
)
st.download_button("Download HTML Report", data=html.encode("utf-8"), file_name="osint_report.html", mime="text/html")
# ---------------------------
# Username Availability Probe (simple)
# ---------------------------
PLATFORM_PATTERNS: Dict[str,str] = {
"GitHub": "https://github.com/{user}",
"Twitter": "https://x.com/{user}",
"Reddit": "https://www.reddit.com/user/{user}",
"Medium": "https://medium.com/@{user}",
}
def probe_usernames(users: List[str], limit: int = 10) -> List[Dict[str,str]]:
out = []
if requests is None:
return out
for u in users[:limit]:
for plat, pattern in PLATFORM_PATTERNS.items():
url = pattern.format(user=u)
status = "?"
try:
r = requests.get(url, timeout=5)
if r.status_code == 200:
status = "Exists"
elif r.status_code == 404:
status = "Not Found"
else:
status = str(r.status_code)
except Exception:
status = "Error"
out.append({"platform": plat, "username": u, "status": status})
return out
def render_step4_execution(entity_type: str, entity_value: str):
st.subheader("Step 4: Execute & Metadata")
final = st.session_state.get("selected_dorks", [])
if not final:
st.info("No dorks selected.")
return
max_per = st.slider("Max results", 3, 20, st.session_state.get("settings", {}).get("max_per", 10))
if st.button("Run dorks"):
# Progressive skeleton loader while executing each query
placeholder = st.empty()
results: List[Dict[str, Any]] = []
total_expected = len(final) * max_per
for i, q in enumerate(final, start=1):
remaining = len(final) - i + 1
est_remaining = remaining * max_per
# Render skeletons representing expected remaining results (capped for performance)
with placeholder.container():
st.markdown("#### Running searches…")
st.caption(f"Query {i}/{len(final)}: {q}")
skel_blocks = min(est_remaining, 18) # avoid huge DOM
# Distribute size variations for visual interest
sizes = ["sm", "md", "lg"]
rows_html = []
for j in range(skel_blocks):
size = sizes[j % len(sizes)]
rows_html.append(f'')
st.markdown(
'' + "".join(rows_html) + "
",
unsafe_allow_html=True,
)
# Execute the actual search
rows = ddg_search(q, max_results=max_per)
_audit_log("dork_run", dork=q, results=len(rows))
results.extend(rows)
# Clear placeholder after completion
placeholder.empty()
st.session_state["dork_results"] = results
# compute scores after acquiring all results
facts = KnownFacts.from_session()
st.session_state["scored_results"] = score_all_findings(results, facts)
if res := st.session_state.get("dork_results"):
st.json(res)
audit_str = "\n".join(json.dumps(ev) for ev in st.session_state["audit"])
st.download_button("Download audit", audit_str, "audit.jsonl")
st.markdown("---")
st.subheader("File/Image Metadata Extractor")
upload = st.file_uploader("Upload a file (pdf, docx, mp3, jpg, etc.)")
if upload:
meta = extract_metadata(upload)
st.json(meta)
# ---------------------------
# Main
# ---------------------------
def render_help_tab():
st.subheader("How To Use This OSINT Investigator Suite")
st.markdown("""
This tab is a quick field manual. It shows the purpose of every tab, the workflow order, and pro tips.
### Recommended Workflow (Fast Path)
1. Known Facts – Load seed identifiers (handles, emails, domains, names).
2. Plan – Review the autogenerated phased investigation plan; adjust facts if gaps obvious.
3. Explainer – Learn the dork building logic for transparency (optional).
4. Advisor – Get recommended dorks (rule + optional LLM). Refine, then accept.
5. Selection – Curate / edit / remove dorks; finalize the set to run.
6. Execution – Run dorks (skeleton loaders show progress); extract file/image metadata if you have artifacts.
7. Scoring – Review confidence scores, filter, read explanations, iterate by adding new facts and re-scoring.
8. Graph – Visual relationship view (requires networkx + pyvis) to spot high‑intersection nodes.
9. Report – Export an HTML snapshot for stakeholders / evidence chain.
10. Usernames – Probe handle existence across common platforms.
11. Help – (This) reference card anytime.
---
### Tab Details & Tips
**Known Facts**
- Add all solid identifiers early; scoring & dork generation leverage them.
- Handles & emails dramatically raise confidence when co-occurring in sources.
- Update facts after each scouting loop (new domains from findings, etc.).
**Plan**
- Generated phases: Recon, Expansion, Correlation, Deep Dive, Reporting.
- Use it as a narrative backbone for your final export or task tickets.
**Explainer**
- Shows how base + contextual tokens assemble into search dorks by entity type.
- Use to justify methodology or teach newcomers.
**Advisor**
- Hybrid: deterministic heuristic rules plus optional LLM (if HF token + model set in settings).
- Toggle embedding/semantic features in settings (if present) to enrich scoring later.
- Accept the generated list to push candidates to Selection.
**Selection**
- Final edit surface. Remove noisy / redundant queries before execution.
- Keep a balanced mix: broad footprint + specific leak/file/resource patterns.
**Execution**
- Click Run dorks: animated skeleton placeholders appear per batch while searches resolve.
- Results cached in session: re-running overwrites (audit log tracks runs).
- Metadata Extractor: Upload docs / images to pull EXIF, PDF metadata, docx core props, audio tags.
**Scoring**
- Each finding scored from component signals (exact identifiers, fuzzy tokens, co-occurrence, reliability, context keywords, semantic similarity).
- Levels: High ≥70, Medium ≥40. Use filters + search bar to triage.
- Re-score after updating Known Facts or enabling embeddings.
- "Full Explanations" expands reasoning transparency for defensibility.
**Graph**
- Visual pivot map: nodes sized by aggregated score; edges for shared identifiers.
- Use to spot central assets (good pivot candidates) quickly.
- If graph libs missing you'll see an install hint (they're listed in requirements).
**Report**
- Generates a standalone HTML (includes styling + key metrics) for sharing.
- Consider exporting after each major iteration to preserve state (version trail).
**Usernames**
- Lightweight existence probe (HTTP status heuristic). "Exists" ≠ ownership proof.
- Add more platforms by extending PLATFORM_PATTERNS in code.
**Chat Assistant (Floating)**
- Noir-style guidance; quick buttons for common pivots.
- If a model + token configured, responses may blend LLM nuance with rule hints; otherwise rule-based only.
- Close with ✕; reopen with the 🕵️ button.
**Light / Dark Toggle**
- Sidebar toggle (if present) swaps theme classes; custom components auto-adapt.
**Skeleton Loaders**
- Shimmering bars appear during long search batches to indicate progress.
---
### Power User Tips
- Iterative Loop: (Run) → (Score) → (Add new facts from findings) → (Re-score) → (Graph) → (Report).
- High-value pivots: Rare email domains, unique handles in code repos, author names in PDF metadata.
- Noise Control: Remove generic dorks that return unrelated trending content before executing.
- Evidence Chain: Audit log (download on Execution tab) + HTML reports form a defensible trail.
### Performance Notes
- Limiting Max results reduces API latency & keeps scoring responsive.
- Embedding model loads lazily—first semantic scoring may pause a few seconds.
- Graph view caps large result sets to avoid browser lockups.
### Glossary
- Dork: Crafted search query combining identifiers + context tokens.
- Pivot: New investigative direction unlocked by a discovered unique attribute.
- Co-occurrence: Multiple target identifiers appearing together in one source.
### Ethics Reminder
Public sources only. No credential stuffing, intrusion, or accessing private data stores. Respect rate limits & platform ToS.
""")
def main():
st.markdown("""
🕵️
OSINT Investigator Suite
AI-Augmented
Heuristic Scoring
Report Export
""", unsafe_allow_html=True)
entity_type = st.selectbox("Entity type", list(TYPED_DORK_MAP.keys()), key="entity_type")
entity_value = st.text_input("Entity value", "user@example.com", key="entity_value")
if entity_type and entity_value:
tabs = st.tabs(["Known Facts", "Plan", "Explainer", "Advisor", "Selection", "Execution", "Scoring", "Graph", "Report", "Usernames", "Help"])
with tabs[0]:
_known_facts_ui()
with tabs[1]:
render_investigation_plan(entity_type, entity_value)
with tabs[2]:
render_dorks_explainer(entity_type, entity_value)
with tabs[3]:
render_dork_recommender(entity_type, entity_value)
with tabs[4]:
render_dork_selection(entity_type, entity_value)
with tabs[5]:
render_step4_execution(entity_type, entity_value)
with tabs[6]:
st.subheader("Scoring & Confidence")
facts = KnownFacts.from_session()
scored = st.session_state.get("scored_results")
if not scored:
st.info("Run dorks first to generate findings and scores.")
else:
high = sum(1 for r in scored if r["level"] == "High")
med = sum(1 for r in scored if r["level"] == "Medium")
low = sum(1 for r in scored if r["level"] == "Low")
st.markdown("Findings Overview
", unsafe_allow_html=True)
k1,k2,k3,k4 = st.columns(4)
k1.metric("Total", len(scored))
k2.metric("High", high)
k3.metric("Medium", med)
k4.metric("Low", low)
level_filter = st.multiselect("Levels", ["High", "Medium", "Low"], default=["High", "Medium", "Low"], key="lvl_filter")
q = st.text_input("Search title/snippet", key="score_search")
view = [r for r in scored if r["level"] in level_filter and (not q or q.lower() in (r.get("snippet", '')).lower() or q.lower() in (r.get("title", '')).lower())]
rows_html = []
for r in view:
lvl = r["level"].lower()
badge = f"{r['level']}"
title = (r.get('title',''))[:120]
expl_short = (r.get('explanation',''))[:180]
url = r.get('url') or ''
rows_html.append(f"| {r['score']} | {badge} | {title} | link | {r['reliability']} | {expl_short} |
")
table_html = """
| Score | Level | Title | URL | Reliab. | Explanation (truncated) |
|---|
{rows}
""".format(rows="".join(rows_html))
st.markdown(table_html, unsafe_allow_html=True)
col_rescore, col_full, col_export = st.columns([1,2,1])
with col_rescore:
if st.button("Re-score", key="btn_rescore_now"):
rescored = score_all_findings(st.session_state.get("dork_results", []), facts)
st.session_state["scored_results"] = rescored
st.success("Re-scored.")
with col_full:
with st.expander("Full Explanations"):
for r in view:
st.markdown(f"**{r.get('title','')}** — {r['level']} ({r['score']})\n\n{r.get('explanation','')}")
with col_export:
if st.button("Export Report (HTML)", key="btn_export_report_inline"):
export_report(entity_type, entity_value, facts, scored)
with tabs[7]:
st.subheader("Entity Graph")
facts = KnownFacts.from_session()
scored = st.session_state.get("scored_results") or []
if scored:
html = build_graph(scored, facts)
if html:
st.components.v1.html(html, height=600, scrolling=True)
else:
st.info("Install networkx & pyvis for graph visualization.")
else:
st.info("No scored findings yet.")
with tabs[8]:
st.subheader("Report Export")
facts = KnownFacts.from_session()
scored = st.session_state.get("scored_results") or []
if scored:
export_report(entity_type, entity_value, facts, scored)
else:
st.info("Run and score findings to export a report.")
with tabs[9]:
st.subheader("Username Availability Probe")
facts = KnownFacts.from_session()
sample_users = facts.handles[:10] or [entity_value] if entity_type == "Username / Handle" else []
if not sample_users:
st.info("Add handles in Known Facts or pick a username entity.")
else:
if st.button("Probe Platforms", key="btn_probe_users"):
data = probe_usernames(sample_users)
st.session_state["probe_results"] = data
if pr := st.session_state.get("probe_results"):
st.dataframe(pr, use_container_width=True)
with tabs[10]:
render_help_tab()
# Floating chat widget render
render_chat_widget(entity_type, entity_value)
with st.expander("Methodology / Scoring Rubric", expanded=False):
st.markdown("""
**Scoring Components**
- Email (+25) / Name exact (+20) / Handle (+15) / Domain (+10) / IP (+10) / Org (+8)
- Fuzzy name token (+8) / Co-occurrence (+10)
- Source reliability High (+10) / Medium (+5)
- Context alignment (1:+3 / 2:+6 / ≥3:+10)
- Semantic similarity (0–20 scaled) if enabled
**Levels:** High ≥70, Medium ≥40, else Low.
""")
with st.expander("Ethical Use Notice", expanded=False):
st.markdown("Lawful OSINT only. No intrusion, auth bypass, or accessing non-public data. Respect platform ToS & privacy.")
# ---------------------------
# Chat Assistant
# ---------------------------
GUIDE_SYSTEM = (
"You are a noir-style seasoned OSINT investigator named 'The Analyst'. Speak like classic crime noir: terse, vivid metaphors, professional, never cheesy. "
"Guide the user step-by-step in enumerating a digital entity using only ethical open sources. "
"Each answer: <=150 words, 2-4 compact paragraphs or bullet fragments. Provide concrete next actions, pivot angles, and a light ethics reminder if user drifts. "
"Avoid sensationalism. No illegal guidance. Occasionally finish with a brief noir tag line like 'That's the shape of the alley, kid.'" )
def _summarize_context(entity_type: str, entity_value: str) -> str:
facts: KnownFacts = KnownFacts.from_session()
scored = st.session_state.get("scored_results") or []
high_titles = [s.get("title") for s in scored if s.get("level") == "High"][:5]
parts = [f"Entity: {entity_type}={entity_value}"]
if facts.handles: parts.append(f"Handles:{len(facts.handles)}")
if facts.emails: parts.append(f"Emails:{len(facts.emails)}")
if facts.domains: parts.append(f"Domains:{len(facts.domains)}")
if high_titles: parts.append("HighHits:" + ";".join(high_titles))
return " | ".join(parts)
def _rule_based_reply(user_msg: str, entity_type: str, entity_value: str) -> str:
msg = user_msg.lower()
lines = []
ctx = _summarize_context(entity_type, entity_value)
if any(k in msg for k in ["start", "hello", "hi", "first"]):
lines.append("First we empty our pockets—handles, domains, emails. Solid identifiers become compass bearings.")
if "dork" in msg or "search" in msg:
lines.append("Open with wide footprint dorks. Then tighten: docs leaks, repo chatter, paste traces. Each query is a flashlight beam.")
if "score" in msg or "confidence" in msg:
lines.append("Confidence breathes when multiple facts collide in a clean source. Add precise emails or stable handles—re-score, watch the highs rise.")
if "graph" in msg:
lines.append("Graph shows the intersections. Nodes struck by multiple identifiers—those corners hide stories.")
if "pivot" in msg or "next" in msg:
lines.append("Pivot off unique anchors: a handle in a PDF, an email in a commit, a domain in a press note. Each pivot narrows the alley.")
if not lines:
lines.append("Playbook: 1) Lock facts 2) Advisor for 10 sharp dorks 3) Select & run 4) Score 5) Add new facts 6) Graph pivots 7) Export report.")
lines.append(f"Context snapshot: {ctx}")
lines.append("Stay clean—public sources only. That's the shape of the alley, kid.")
return "\n\n".join(lines)
def render_chat_widget(entity_type: str, entity_value: str):
# Session setup
st.session_state.setdefault("chat_history", [])
st.session_state.setdefault("chat_open", True)
open_flag = st.session_state["chat_open"]
# Mini open button (when closed)
if not open_flag:
if st.button("🕵️", key="open_chat_button"):
st.session_state["chat_open"] = True
# Style the button to float
st.markdown("""
""", unsafe_allow_html=True)
return
# Build chat window
messages = st.session_state["chat_history"]
# Render HTML shell
st.markdown("", unsafe_allow_html=True)
# Header with close control
c1, c2, c3 = st.columns([0.2, 0.65, 0.15])
with c1:
st.markdown("", unsafe_allow_html=True)
with c2:
st.markdown("", unsafe_allow_html=True)
with c3:
if st.button("✕", key="close_chat_btn"):
st.session_state["chat_open"] = False
st.markdown("
", unsafe_allow_html=True)
return
# Messages area
# Use an empty container to emulate scroll (Streamlit limitation)
msg_container = st.container()
with msg_container:
if messages:
for turn in messages[-18:]:
st.markdown(f"You: {turn['user']}
", unsafe_allow_html=True)
st.markdown(f"Inv: {turn['assistant']}
", unsafe_allow_html=True)
else:
st.markdown("Need a lead? Ask me about dorks, scoring, or pivots.
", unsafe_allow_html=True)
# Input form
with st.form("chat_form", clear_on_submit=True):
q = st.text_area("Message", key="chat_input_area", height=70, label_visibility="collapsed")
col_a, col_b, col_c, col_d = st.columns(4)
send = False
with col_a:
if st.form_submit_button("Send"):
send = True
with col_b:
if st.form_submit_button("Dorks"):
q = "What dorks should I run next?"; send = True
with col_c:
if st.form_submit_button("Confidence"):
q = "How do I improve confidence now?"; send = True
with col_d:
if st.form_submit_button("Pivot"):
q = "Give me a pivot strategy."; send = True
if send and q.strip():
reply: Optional[str] = None
if st.session_state.get("settings", {}).get("model") and os.getenv("HF_API_TOKEN"):
convo = st.session_state["chat_history"][-6:]
history_str = "\n".join([f"User: {h['user']}\nAssistant: {h['assistant']}" for h in convo if h.get('assistant')])
prompt = (
f"{GUIDE_SYSTEM}\nCurrentContext: {_summarize_context(entity_type, entity_value)}\n" +
history_str + f"\nUser: {q}\nAssistant:")
reply = _hf_infer(MODEL_ID_MAP.get(st.session_state["settings"]["model"], st.session_state["settings"]["model"]), prompt, max_new_tokens=190, temperature=0.35)
if not reply:
reply = _rule_based_reply(q, entity_type, entity_value)
st.session_state["chat_history"].append({"user": q, "assistant": reply})
st.markdown("Ethical OSINT only.🕵️♂️
", unsafe_allow_html=True)
st.markdown("", unsafe_allow_html=True)
if __name__ == "__main__":
main()