Mythos / app.py
Amrmostafa25's picture
Add live domain security scanner with streaming results and strip model reasoning
f17e2a4 verified
Raw
History Blame Contribute Delete
35.1 kB
import os
import re
import socket
import ssl
import time
from datetime import datetime, timezone
from urllib.error import URLError
from urllib.request import Request, urlopen
import gradio as gr
from huggingface_hub import InferenceClient
INFERENCE_MODEL_ID = os.getenv(
"INFERENCE_MODEL_ID",
"empero-ai/Qwythos-9B-Claude-Mythos-5-1M",
)
DEFAULT_SYSTEM = (
"You are Mythos, an elite cybersecurity AI assistant. "
"Output ONLY the final answer in markdown. "
"NEVER output reasoning, planning, or phrases like 'The user wants' or 'I need to'. "
"If scan results are provided, summarize vulnerabilities with severity and fixes. "
"Match the user's language. Be precise and actionable."
)
QUICK_ACTIONS = {
"🔍 OWASP Scan": "Run full OWASP security scan on yexa.pro and list all findings with severity.",
"🌐 API Security": "Scan yexa.pro API endpoints and list security risks with mitigations.",
"🧪 JWT Audit": "Audit JWT and authentication security for yexa.pro.",
"📦 Headers Scan": "Check security headers and TLS configuration for yexa.pro.",
"📄 Full Report": "Generate executive security report for yexa.pro based on live scan.",
"🛡️ RBAC Review": "Review access control and exposure surface for yexa.pro.",
}
DOMAIN_RE = re.compile(
r"(?:https?://)?(?:www\.)?"
r"([a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?"
r"(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)+)",
re.IGNORECASE,
)
SCAN_TRIGGER_RE = re.compile(
r"\b(test|scan|check|audit|pentest|فحص|اختبر|اختبار|افحص|تحقق|امسح)\b",
re.IGNORECASE,
)
REASONING_PREFIXES = (
"the user wants",
"i need to",
"i should",
"i will",
"i'll",
"let me",
"this is a",
"the best way",
"i can use",
"i'm going to",
"okay,",
"since i",
)
CUSTOM_CSS = """
@import url('https://fonts.googleapis.com/css2?family=IBM+Plex+Sans+Arabic:wght@400;500;600;700&family=JetBrains+Mono:wght@400;600&family=Inter:wght@400;500;600;700&display=swap');
:root {
--bg: #05080d;
--panel: #0b1220;
--panel-2: #101a2b;
--line: #1f3b34;
--green: #10b981;
--green-soft: #34d399;
--text: #ecfdf5;
--muted: #94a3b8;
}
html, body {
overflow-x: hidden !important;
width: 100% !important;
}
.gradio-container {
max-width: 100% !important;
width: 100% !important;
margin: 0 auto !important;
padding: 0.5rem 1rem 1rem !important;
font-family: 'IBM Plex Sans Arabic', 'Inter', sans-serif !important;
background: radial-gradient(circle at 20% 0%, #0a1a14 0%, var(--bg) 45%) !important;
}
.gradio-container .contain {
max-width: 100% !important;
}
.gradio-container .block,
.gradio-container .form,
.gradio-container .panel {
background: transparent !important;
border-color: #1f3b34 !important;
}
.gradio-container .mythos-chatbot,
.gradio-container .mythos-chatbot.block,
.gradio-container .mythos-chatbot .block,
.gradio-container .mythos-chatbot .wrap {
background: #070d14 !important;
}
.gradio-container .block-label,
.gradio-container label span {
color: #cbd5e1 !important;
}
.mythos-topbar {
display: flex; justify-content: space-between; align-items: center;
background: linear-gradient(90deg, #07130f, #0b1a2f);
border: 1px solid var(--line);
border-radius: 14px;
padding: 0.9rem 1.2rem;
margin-bottom: 0.8rem;
}
.mythos-topbar .brand { display: flex; gap: 0.75rem; align-items: center; }
.mythos-logo {
width: 42px; height: 42px; border-radius: 10px;
background: linear-gradient(135deg, #059669, #22d3ee);
display: grid; place-items: center; font-weight: 800; color: #04120d;
box-shadow: 0 0 20px rgba(16,185,129,.35);
}
.mythos-topbar h1 { margin: 0; color: var(--text); font-size: 1.35rem; }
.mythos-topbar p { margin: 0; color: var(--muted); font-size: 0.82rem; }
.mythos-status { display: flex; gap: 0.5rem; flex-wrap: wrap; justify-content: flex-end; }
.status-pill {
font-size: 0.72rem; padding: 0.28rem 0.6rem; border-radius: 999px;
border: 1px solid var(--line); color: #a7f3d0; background: #0f1f2f;
font-family: 'JetBrains Mono', monospace;
}
.gradio-container .contain > .mythos-shell,
.gradio-container .mythos-shell {
display: flex !important;
flex-direction: row !important;
flex-wrap: nowrap !important;
align-items: flex-start !important;
gap: 0.8rem !important;
width: 100% !important;
height: auto !important;
}
.gradio-container .mythos-shell > .column,
.gradio-container .mythos-shell > .gap {
height: auto !important;
min-height: 0 !important;
align-self: flex-start !important;
}
.gradio-container .mythos-shell > .column.mythos-sidebar,
.gradio-container .mythos-sidebar {
flex: 0 0 240px !important;
width: 240px !important;
min-width: 240px !important;
max-width: 240px !important;
}
.gradio-container .mythos-shell > .column.mythos-main,
.gradio-container .mythos-main {
flex: 1 1 auto !important;
min-width: 0 !important;
width: auto !important;
}
.mythos-sidebar {
background: var(--panel); border: 1px solid var(--line);
border-radius: 14px; padding: 0.8rem;
align-self: flex-start !important;
}
.mythos-sidebar .block,
.mythos-sidebar .wrap,
.mythos-sidebar fieldset {
background: #0b1220 !important;
border-color: #1f3b34 !important;
box-shadow: none !important;
}
.mythos-sidebar .gr-radio label,
.mythos-sidebar .gr-radio .wrap {
background: #0d1f1a !important;
color: #ecfdf5 !important;
border: 1px solid #1f3b34 !important;
}
.mythos-sidebar .gr-radio input:checked + label,
.mythos-sidebar .gr-radio .selected {
background: linear-gradient(135deg, #064e3b, #0e7490) !important;
border-color: #34d399 !important;
color: #ecfdf5 !important;
box-shadow: 0 0 12px rgba(16,185,129,.2) !important;
}
.mythos-sidebar .gr-button,
.mythos-sidebar button {
background: #0d1f1a !important;
border: 1px solid #1f4d3f !important;
color: #bbf7d0 !important;
}
.mythos-divider {
height: 1px; background: #1f3b34; margin: 0.75rem 0;
}
.mythos-main {
background: var(--panel); border: 1px solid var(--line);
border-radius: 14px; padding: 0.75rem;
max-width: 100% !important;
height: auto !important;
min-height: 0 !important;
align-self: flex-start !important;
}
.chat-panel {
width: 100% !important;
display: block !important;
height: auto !important;
}
.chat-panel > .gap {
display: flex !important;
flex-direction: column !important;
gap: 0.5rem !important;
height: auto !important;
}
.quick-chips {
display: flex !important;
flex-wrap: wrap !important;
gap: 0.4rem !important;
margin: 0.5rem 0 !important;
}
.quick-chips button {
border: 1px solid #1f4d3f !important;
background: #0d1f1a !important;
color: #bbf7d0 !important;
border-radius: 999px !important;
font-size: 0.76rem !important;
padding: 0.35rem 0.7rem !important;
}
.quick-chips button:hover {
border-color: #34d399 !important;
box-shadow: 0 0 12px rgba(16,185,129,.2) !important;
}
.quick-grid { display: grid; grid-template-columns: repeat(3, 1fr); gap: 0.5rem; margin: 0.6rem 0 0.8rem; }
.metric-card {
background: var(--panel-2); border: 1px solid var(--line); border-radius: 12px; padding: 0.8rem;
}
.metric-card .num { font-size: 1.5rem; font-weight: 700; color: var(--green-soft); }
.metric-card .lbl { color: var(--muted); font-size: 0.78rem; margin-top: 0.2rem; }
.metric-card.critical .num { color: #f87171; }
.metric-card.high .num { color: #fb923c; }
.metrics-grid { display: grid; grid-template-columns: repeat(4, 1fr); gap: 0.7rem; margin-bottom: 0.8rem; }
.quick-grid button {
border: 1px solid #1f4d3f !important; background: #0d1f1a !important;
color: #bbf7d0 !important; border-radius: 10px !important;
}
.quick-grid button:hover {
border-color: #34d399 !important; box-shadow: 0 0 16px rgba(16,185,129,.25) !important;
}
.mythos-chatbot .chatbot-placeholder,
.mythos-chatbot .placeholder {
color: #6ee7b7 !important;
padding: 2rem 1rem !important;
text-align: center !important;
font-size: 0.95rem !important;
}
.chat-panel .chatbot,
.mythos-chatbot,
.mythos-chatbot > .wrap,
.mythos-chatbot .wrap,
.mythos-chatbot .form,
.mythos-chatbot .panel,
.mythos-chatbot .message-wrap,
.mythos-chatbot .component-wrap,
.mythos-chatbot .scroll-hide,
.mythos-chatbot .empty,
.mythos-chatbot .chatbot-placeholder {
background: #070d14 !important;
background-color: #070d14 !important;
color: #d9f99d !important;
}
.mythos-chatbot .chatbot-placeholder,
.mythos-chatbot .placeholder {
color: #6ee7b7 !important;
padding: 2rem 1rem !important;
text-align: center !important;
font-size: 0.95rem !important;
}
.mythos-chatbot {
--body-text-color: #d9f99d !important;
--block-background-fill: #070d14 !important;
--background-fill-primary: #070d14 !important;
--background-fill-secondary: #0f1b2d !important;
border: 1px solid #1e3a5f !important;
border-radius: 12px !important;
height: 420px !important;
min-height: 420px !important;
max-height: 420px !important;
flex: 0 0 420px !important;
}
.mythos-chatbot > .wrap {
height: 420px !important;
max-height: 420px !important;
}
.mythos-chatbot .scroll-hide,
.mythos-chatbot .message-wrap {
overscroll-behavior: contain !important;
}
.chat-panel .chatbot {
border: 1px solid #1e3a5f !important;
border-radius: 12px !important;
background: #070d14 !important;
}
.chat-panel .chatbot .label-wrap,
.chat-panel .chatbot > label,
.chat-panel .chatbot legend,
.chat-panel .chatbot .block-label,
.chat-panel .chatbot span[data-testid="block-info"],
.mythos-chatbot .label-wrap,
.mythos-chatbot .block-label {
display: none !important;
height: 0 !important; margin: 0 !important; padding: 0 !important;
}
.chat-panel .chatbot .wrap,
.mythos-chatbot .wrap {
border: none !important;
padding: 0.5rem !important;
}
.mythos-chatbot .message {
margin: 0.5rem 0 !important;
padding: 0.75rem 1rem !important;
border-radius: 12px !important;
max-width: 92% !important;
}
.mythos-chatbot .message.user,
.mythos-chatbot .message.assistant,
.mythos-chatbot .message.bot,
.chat-panel .chatbot .message-wrap .message.user,
.chat-panel .chatbot .message-wrap .message.assistant,
.chat-panel .chatbot .message-wrap .message.bot {
display: block !important;
visibility: visible !important;
opacity: 1 !important;
position: relative !important;
z-index: 2 !important;
}
.mythos-chatbot .message.user,
.chat-panel .chatbot .message-wrap .message.user {
background: linear-gradient(135deg, #047857, #0e7490) !important;
color: #ecfdf5 !important;
border: 1px solid #34d399 !important;
margin-left: auto !important;
}
.mythos-chatbot .message.assistant,
.mythos-chatbot .message.bot,
.chat-panel .chatbot .message-wrap .message.assistant,
.chat-panel .chatbot .message-wrap .message.bot {
background: #0f1b2d !important;
color: #d9f99d !important;
border: 1px solid #1f9d74 !important;
margin-right: auto !important;
}
.mythos-chatbot:has(.message) .placeholder,
.mythos-chatbot:has(.message) .chatbot-placeholder,
.mythos-chatbot:has(.message) .empty {
display: none !important;
}
.mythos-chatbot .message .md,
.mythos-chatbot .message .prose,
.mythos-chatbot .message .prose *,
.mythos-chatbot .message p,
.mythos-chatbot .message li,
.mythos-chatbot .message span,
.mythos-chatbot .message strong,
.mythos-chatbot .message em,
.mythos-chatbot .message h1,
.mythos-chatbot .message h2,
.mythos-chatbot .message h3,
.mythos-chatbot .message code,
.mythos-chatbot .message pre,
.chat-panel .chatbot .message-wrap .message.user .prose,
.chat-panel .chatbot .message-wrap .message.user .prose *,
.chat-panel .chatbot .message-wrap .message.assistant .prose,
.chat-panel .chatbot .message-wrap .message.assistant .prose * {
color: inherit !important;
background: transparent !important;
}
.mythos-chatbot .message.user .md,
.mythos-chatbot .message.user .prose,
.mythos-chatbot .message.user .prose * {
color: #ecfdf5 !important;
}
.mythos-chatbot .message.assistant .md,
.mythos-chatbot .message.assistant .prose,
.mythos-chatbot .message.assistant .prose * {
color: #d9f99d !important;
}
.mythos-chatbot .message pre,
.mythos-chatbot .message code {
background: #0a1520 !important;
border: 1px solid #1f3b34 !important;
color: #86efac !important;
}
.mythos-chatbot .avatar-container,
.chat-panel .chatbot .avatar-container { display: none !important; }
.mythos-chatbot .icon-button,
.mythos-chatbot .copy-btn {
background: #0d1f1a !important;
border: 1px solid #1f4d3f !important;
color: #bbf7d0 !important;
}
.chat-input-row {
align-items: flex-end !important;
gap: 0.6rem !important;
margin-top: 0.5rem !important;
}
.chat-input-row > .block,
.chat-input-row > .form,
.chat-input-row .wrap {
margin: 0 !important;
background: transparent !important;
border: none !important;
box-shadow: none !important;
}
.chat-input-row textarea {
border: 1px solid #1f9d74 !important;
background: #0b1522 !important;
color: var(--text) !important;
border-radius: 12px !important;
min-height: 64px !important;
}
.chat-input-row button {
min-height: 64px !important;
min-width: 110px !important;
border-radius: 12px !important;
background: linear-gradient(135deg, #059669, #047857) !important;
border: 1px solid #34d399 !important;
color: #ecfdf5 !important;
font-weight: 600 !important;
}
.chat-input-row button:hover {
box-shadow: 0 0 18px rgba(16,185,129,.35) !important;
}
.chat-panel textarea {
border: 1px solid #1f9d74 !important; background: #0b1522 !important;
color: var(--text) !important; border-radius: 12px !important;
}
.mythos-sidebar .status-box {
background: #0a141f !important;
border: 1px solid #1f9d74 !important;
border-radius: 10px !important;
padding: 0.65rem 0.75rem !important;
text-align: center !important;
font-family: 'JetBrains Mono', monospace !important;
font-size: 0.88rem !important;
color: #34d399 !important;
margin-top: 0.5rem !important;
}
.mythos-sidebar .status-box p { margin: 0 !important; color: #34d399 !important; }
.empty-state {
border: 1px dashed #1f9d74; border-radius: 12px; padding: 1rem;
color: #a7f3d0; text-align: center; margin-bottom: 0.7rem; background: #0a141f;
}
footer { display: none !important; }
@media (max-width: 980px) {
.gradio-container .mythos-shell {
flex-direction: column !important;
}
.gradio-container .mythos-shell > .column.mythos-sidebar,
.gradio-container .mythos-sidebar {
flex: 1 1 auto !important;
width: 100% !important;
min-width: 0 !important;
max-width: 100% !important;
}
.metrics-grid { grid-template-columns: repeat(2, 1fr); }
.quick-grid { grid-template-columns: 1fr; }
}
"""
THEME = gr.themes.Base(
primary_hue=gr.themes.colors.emerald,
secondary_hue=gr.themes.colors.cyan,
neutral_hue=gr.themes.colors.slate,
font=gr.themes.GoogleFont("Inter"),
).set(
body_background_fill="#05080d",
block_background_fill="#0b1220",
block_border_color="#1f3b34",
block_label_text_color="#cbd5e1",
body_text_color="#ecfdf5",
input_background_fill="#0b1522",
input_border_color="#1f3b34",
button_secondary_background_fill="#0d1f1a",
button_secondary_text_color="#bbf7d0",
button_primary_background_fill="#059669",
button_primary_background_fill_hover="#047857",
block_title_text_color="#ecfdf5",
body_text_color_subdued="#94a3b8",
)
DASHBOARD_HTML = """
<div class="metrics-grid">
<div class="metric-card critical"><div class="num">12</div><div class="lbl">Critical Findings</div></div>
<div class="metric-card high"><div class="num">37</div><div class="lbl">High Risk</div></div>
<div class="metric-card"><div class="num">74</div><div class="lbl">Medium</div></div>
<div class="metric-card"><div class="num">8.4</div><div class="lbl">Risk Score</div></div>
</div>
<div class="empty-state"><strong>Security Operations Overview</strong><br/>Live scans and agent status appear here.</div>
"""
REPORTS_HTML = """
<div class="empty-state"><strong>Reports Center</strong><br/>Generate Executive / Technical / OWASP reports.</div>
"""
def extract_domain(text: str) -> str | None:
match = DOMAIN_RE.search(text or "")
if not match:
return None
domain = match.group(1).lower().rstrip(".")
if domain in {"example.com", "localhost"}:
return None
return domain
def should_run_live_scan(text: str) -> bool:
domain = extract_domain(text)
if not domain:
return False
return bool(SCAN_TRIGGER_RE.search(text)) or "scan" in text.lower() or "test" in text.lower()
def _fetch(url: str, timeout: int = 10):
request = Request(url, headers={"User-Agent": "Mythos-Security-Scanner/1.0"})
with urlopen(request, timeout=timeout) as response:
body = response.read(12_000)
headers = {k.lower(): v for k, v in response.headers.items()}
return response.status, headers, body
def _severity_badge(level: str) -> str:
icons = {"critical": "🔴", "high": "🟠", "medium": "🟡", "low": "🔵", "info": "⚪"}
return f"{icons.get(level, '⚪')} **{level.upper()}**"
def run_live_security_scan(domain: str):
findings: list[tuple[str, str, str]] = []
lines: list[str] = [
f"## 🛡️ Mythos Live Security Scan",
f"**Target:** `{domain}` ",
f"**Started:** {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}",
"",
]
yield "\n".join(lines)
lines += ["### 🟡 Phase 1/5 — DNS", ""]
yield "\n".join(lines)
try:
records = sorted({item[4][0] for item in socket.getaddrinfo(domain, None)})
lines.append(f"✅ Resolved **{len(records)}** IP(s): `{', '.join(records[:6])}`")
except socket.gaierror as exc:
lines.append(f"❌ DNS failed: `{exc}`")
findings.append(("high", "DNS", f"Domain does not resolve: {exc}"))
yield "\n".join(lines)
lines += ["", "### 🟡 Phase 2/5 — TLS / HTTPS", ""]
yield "\n".join(lines)
try:
context = ssl.create_default_context()
with socket.create_connection((domain, 443), timeout=10) as sock:
with context.wrap_socket(sock, server_hostname=domain) as secure:
cert = secure.getpeercert()
protocol = secure.version() or "unknown"
subject = dict(x[0] for x in cert.get("subject", ()))
issuer = dict(x[0] for x in cert.get("issuer", ()))
not_after = cert.get("notAfter", "unknown")
lines.append(f"✅ TLS **{protocol}** — cert for `{subject.get('commonName', domain)}`")
lines.append(f"• Issuer: `{issuer.get('commonName', 'unknown')}`")
lines.append(f"• Expires: `{not_after}`")
if protocol in {"TLSv1", "TLSv1.1"}:
findings.append(("high", "TLS", f"Weak protocol: {protocol}"))
except Exception as exc:
lines.append(f"❌ HTTPS/TLS error: `{exc}`")
findings.append(("high", "TLS", f"HTTPS not available or invalid: {exc}"))
yield "\n".join(lines)
lines += ["", "### 🟡 Phase 3/5 — HTTP Probe", ""]
yield "\n".join(lines)
http_status = None
https_status = None
response_headers: dict[str, str] = {}
final_url = f"https://{domain}/"
try:
https_status, response_headers, _ = _fetch(final_url)
lines.append(f"✅ `GET https://{domain}/` → **{https_status}**")
except URLError:
try:
http_status, response_headers, _ = _fetch(f"http://{domain}/")
final_url = f"http://{domain}/"
lines.append(f"⚠️ HTTPS failed; HTTP responds **{http_status}**")
findings.append(("medium", "Transport", "Site reachable over HTTP but HTTPS failed"))
except URLError as exc:
lines.append(f"❌ HTTP probe failed: `{exc}`")
findings.append(("high", "Availability", f"Web probe failed: {exc}"))
if http_status is not None and https_status is None:
findings.append(("high", "Transport", "No HTTPS — traffic may be unencrypted"))
yield "\n".join(lines)
lines += ["", "### 🟡 Phase 4/5 — Security Headers (OWASP)", ""]
yield "\n".join(lines)
required_headers = {
"strict-transport-security": ("medium", "Missing HSTS header"),
"content-security-policy": ("medium", "Missing Content-Security-Policy"),
"x-frame-options": ("low", "Missing X-Frame-Options (clickjacking risk)"),
"x-content-type-options": ("low", "Missing X-Content-Type-Options"),
"referrer-policy": ("info", "Missing Referrer-Policy"),
"permissions-policy": ("info", "Missing Permissions-Policy"),
}
for header, (level, message) in required_headers.items():
if header in response_headers:
lines.append(f"✅ `{header}`: `{response_headers[header][:80]}`")
else:
lines.append(f"⚠️ Missing `{header}`")
findings.append((level, "Headers", message))
server = response_headers.get("server", "")
if server:
lines.append(f"• `Server` header exposed: `{server}`")
findings.append(("low", "Fingerprinting", f"Server header disclosed: {server}"))
powered = response_headers.get("x-powered-by", "")
if powered:
lines.append(f"• `X-Powered-By` exposed: `{powered}`")
findings.append(("low", "Fingerprinting", f"X-Powered-By disclosed: {powered}"))
yield "\n".join(lines)
lines += ["", "### 🟡 Phase 5/5 — Surface Checks", ""]
yield "\n".join(lines)
paths = ["/robots.txt", "/.well-known/security.txt", "/api", "/swagger", "/.env", "/admin"]
for path in paths:
url = f"{final_url.rstrip('/')}{path}"
try:
status, _, _ = _fetch(url, timeout=6)
if status < 400:
level = "high" if path in {"/.env", "/admin"} else "medium" if path in {"/api", "/swagger"} else "info"
lines.append(f"{'🔴' if level == 'high' else '🟡'} `{path}` → **{status}** (accessible)")
if level in {"high", "medium"}:
findings.append((level, "Exposure", f"{path} returned HTTP {status}"))
else:
lines.append(f"✅ `{path}` → {status}")
except URLError:
lines.append(f"✅ `{path}` → not reachable")
yield "\n".join(lines)
time.sleep(0.05)
counts = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0}
for level, _, _ in findings:
counts[level] = counts.get(level, 0) + 1
risk = min(10.0, counts["critical"] * 3 + counts["high"] * 2 + counts["medium"] * 1.2 + counts["low"] * 0.5)
lines += [
"",
"### 📊 Scan Summary",
f"• Risk Score: **{risk:.1f}/10**",
f"• Findings: 🔴 {counts['critical']} critical · 🟠 {counts['high']} high · "
f"🟡 {counts['medium']} medium · 🔵 {counts['low']} low",
"",
"### 🚨 Findings",
]
if not findings:
lines.append("✅ No major issues detected in passive checks.")
else:
for level, category, detail in findings:
lines.append(f"- {_severity_badge(level)} **{category}** — {detail}")
lines += [
"",
"_Passive scan only. Do not scan targets you do not own or lack permission to test._",
]
yield "\n".join(lines)
return "\n".join(lines)
def clean_response(text: str) -> str:
if not text:
return text
cleaned = text
for pattern in (
r"<think(?:ing)?>.*?</think(?:ing)?>\s*",
r"<think>.*?</think>\s*",
r"``\s*",
):
cleaned = re.sub(pattern, "", cleaned, flags=re.DOTALL | re.IGNORECASE)
lines = [line.strip() for line in cleaned.splitlines() if line.strip()]
filtered = [
line for line in lines
if not any(line.lower().startswith(prefix) for prefix in REASONING_PREFIXES)
]
if len(lines) > 8 and len(filtered) < max(3, len(lines) // 3):
code_blocks = re.findall(r"```[\s\S]*?```", cleaned)
if code_blocks:
return "\n\n".join(code_blocks).strip()
return "\n".join(filtered[-5:]).strip() if filtered else cleaned[:800].strip()
repetitive = cleaned.count("I'll also include") + cleaned.count("the default is fine")
if repetitive >= 3:
return "\n".join(filtered[-4:]).strip() if filtered else "⚠️ تم تقليص رد مكرر من النموذج. أعد المحاولة بسؤال أقصر."
return cleaned.strip()
def stream_chat(message, history, system_message, max_tokens, temperature, top_p, hf_token: gr.OAuthToken | None = None):
history = list(history or [])
if not message or not message.strip():
yield history, "", "🟢 Ready"
return
user_text = message.strip()
history = history + [{"role": "user", "content": user_text}]
yield history, "", "🟡 Thinking"
if hf_token is None or not getattr(hf_token, "token", None):
history = history + [{
"role": "assistant",
"content": "🔐 سجّل الدخول من زر Hugging Face في الشريط الجانبي ثم أعد المحاولة.",
}]
yield history, "", "🔴 Auth Required"
return
domain = extract_domain(user_text)
if domain and should_run_live_scan(user_text):
history = history + [{"role": "assistant", "content": f"🟡 بدء الفحص الحي لـ `{domain}`..."}]
yield history, "", "🟡 Scanning"
scan_report = ""
for partial in run_live_security_scan(domain):
scan_report = partial
history = history[:-1] + [{"role": "assistant", "content": partial}]
yield history, "", "🟡 Scanning"
client = InferenceClient(token=hf_token.token, model=INFERENCE_MODEL_ID)
analysis_prompt = (
f"{system_message}\n\n"
"Analyze these LIVE scan results. Output ONLY:\n"
"1) Executive summary (3 lines)\n"
"2) Top 5 prioritized fixes\n"
"3) OWASP mapping\n"
"No reasoning text.\n\n"
f"SCAN RESULTS:\n{scan_report}"
)
api_messages = [
{"role": "system", "content": analysis_prompt},
{"role": "user", "content": f"Summarize security findings for {domain}"},
]
history = history[:-1] + [{
"role": "assistant",
"content": scan_report + "\n\n---\n\n### 🤖 Mythos AI Analysis\n⏳ جاري التحليل...",
}]
yield history, "", "🟣 Analyzing"
raw = ""
try:
for chunk in client.chat_completion(
api_messages,
max_tokens=min(int(max_tokens), 1024),
stream=True,
temperature=float(temperature),
top_p=float(top_p),
):
if chunk.choices and chunk.choices[0].delta.content:
raw += chunk.choices[0].delta.content
analysis = clean_response(raw) or "⏳ جاري التحليل..."
history = history[:-1] + [{
"role": "assistant",
"content": scan_report + "\n\n---\n\n### 🤖 Mythos AI Analysis\n" + analysis,
}]
yield history, "", "🟣 Generating"
if not raw.strip():
history = history[:-1] + [{
"role": "assistant",
"content": scan_report + "\n\n---\n\n✅ الفحص الحي اكتمل.",
}]
yield history, "", "🟢 Ready"
except Exception as exc:
history = history[:-1] + [{
"role": "assistant",
"content": scan_report + f"\n\n---\n\n⚠️ تحليل AI فشل: `{exc}`",
}]
yield history, "", "🔴 Error"
return
client = InferenceClient(token=hf_token.token, model=INFERENCE_MODEL_ID)
api_messages = [{"role": "system", "content": system_message}]
api_messages.extend(history)
history = history + [{"role": "assistant", "content": "⏳ جاري الكتابة..."}]
yield history, "", "🟣 Generating"
raw = ""
try:
for chunk in client.chat_completion(
api_messages,
max_tokens=int(max_tokens),
stream=True,
temperature=float(temperature),
top_p=float(top_p),
):
if chunk.choices and chunk.choices[0].delta.content:
raw += chunk.choices[0].delta.content
reply = clean_response(raw) or "⏳ جاري الكتابة..."
history = history[:-1] + [{"role": "assistant", "content": reply}]
yield history, "", "🟣 Generating"
if not raw.strip():
history = history[:-1] + [{
"role": "assistant",
"content": "⚠️ لم يصل رد من النموذج. جرّب: `test example.com` لفحص حي.",
}]
yield history, "", "🟢 Ready"
except Exception as exc:
history = history[:-1] + [{"role": "assistant", "content": f"⚠️ خطأ: {exc}"}]
yield history, "", "🔴 Error"
def switch_page(page):
return (
gr.update(visible=page == "Dashboard"),
gr.update(visible=page == "AI Chat"),
gr.update(visible=page == "Reports"),
gr.update(visible=page == "Settings"),
)
with gr.Blocks(theme=THEME, css=CUSTOM_CSS, title="Mythos") as app:
gr.HTML(
"""
<div class="mythos-topbar">
<div class="brand">
<div class="mythos-logo">M</div>
<div>
<h1>Mythos Security AI</h1>
<p>Qwythos-9B · Cyber Reasoning Platform</p>
</div>
</div>
<div class="mythos-status">
<span class="status-pill">MODEL: QWYTHOS-9B</span>
<span class="status-pill">AGENT: MYTHOS</span>
<span class="status-pill">MODE: INFERENCE API</span>
</div>
</div>
"""
)
system_message = gr.Textbox(value=DEFAULT_SYSTEM, visible=False)
max_tokens = gr.Slider(64, 2048, value=768, step=64, visible=False)
temperature = gr.Slider(0.1, 1.0, value=0.5, step=0.05, visible=False)
top_p = gr.Slider(0.1, 1.0, value=0.9, step=0.05, visible=False)
with gr.Row(elem_classes="mythos-shell"):
with gr.Column(scale=0, min_width=240, elem_classes="mythos-sidebar"):
nav = gr.Radio(
choices=["Dashboard", "AI Chat", "Reports", "Settings"],
value="AI Chat",
label="Navigation",
container=False,
)
gr.HTML('<div class="mythos-divider"></div>')
gr.LoginButton()
gr.Markdown("**Agent Status**", container=False)
status_box = gr.Markdown("🟢 Ready", elem_classes="status-box", container=False)
with gr.Column(scale=1, min_width=320, elem_classes="mythos-main"):
with gr.Group(visible=False) as page_dashboard:
gr.Markdown("## 🛡️ Security Dashboard")
gr.HTML(DASHBOARD_HTML)
with gr.Group(visible=True, elem_classes="chat-panel") as page_chat:
chatbot = gr.Chatbot(
type="messages",
height=420,
show_copy_button=True,
show_label=False,
label="",
avatar_images=(None, None),
latex_delimiters=[],
render_markdown=True,
bubble_full_width=False,
elem_classes="mythos-chatbot",
placeholder="🛡️ اكتب: test yexa.pro — لفحص حي مع بث النتائج والثغرات.",
autoscroll=True,
)
with gr.Row(elem_classes="quick-chips"):
qa_buttons = [gr.Button(label, size="sm") for label in QUICK_ACTIONS]
with gr.Row(elem_classes="chat-input-row"):
msg = gr.Textbox(
placeholder="اكتب طلبك الأمني هنا...",
lines=2,
scale=8,
show_label=False,
)
send = gr.Button("إرسال ➤", variant="primary", scale=1)
with gr.Group(visible=False) as page_reports:
gr.Markdown("## 📄 Reports")
gr.HTML(REPORTS_HTML)
with gr.Group(visible=False) as page_settings:
gr.Markdown("## ⚙️ Advanced Settings")
settings_system = gr.Textbox(value=DEFAULT_SYSTEM, label="System Prompt", lines=4)
settings_max = gr.Slider(64, 2048, value=768, step=64, label="Max Tokens")
settings_temp = gr.Slider(0.1, 1.0, value=0.5, step=0.05, label="Temperature")
settings_top = gr.Slider(0.1, 1.0, value=0.9, step=0.05, label="Top-p")
nav.change(
switch_page,
inputs=[nav],
outputs=[page_dashboard, page_chat, page_reports, page_settings],
)
chat_event_inputs = [msg, chatbot, system_message, max_tokens, temperature, top_p]
chat_event_outputs = [chatbot, msg, status_box]
send.click(stream_chat, inputs=chat_event_inputs, outputs=chat_event_outputs)
msg.submit(stream_chat, inputs=chat_event_inputs, outputs=chat_event_outputs)
for btn, prompt in zip(qa_buttons, QUICK_ACTIONS.values()):
btn.click(lambda p=prompt: p, outputs=msg).then(
stream_chat,
inputs=chat_event_inputs,
outputs=chat_event_outputs,
)
settings_system.change(lambda v: v, inputs=settings_system, outputs=system_message)
settings_max.change(lambda v: v, inputs=settings_max, outputs=max_tokens)
settings_temp.change(lambda v: v, inputs=settings_temp, outputs=temperature)
settings_top.change(lambda v: v, inputs=settings_top, outputs=top_p)
if __name__ == "__main__":
app.launch()