Create app.py

app.py

@@ -0,0 +1,590 @@
+import os
+import re
+import json
+import asyncio
+import ipaddress
+import urllib.parse
+from typing import List, Dict, Any, Optional
+
+import httpx
+from bs4 import BeautifulSoup
+from pydantic import BaseModel, Field, field_validator
+from openai import AsyncOpenAI
+import gradio as gr
+
+# =====================================================================
+# 1. CONFIGURATION & CREDENTIAL MANAGEMENT
+# =====================================================================
+
+DEFAULT_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "")
+DEFAULT_BASE_URL = os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com")
+DEFAULT_MODEL = os.environ.get("DEEPSEEK_MODEL", "deepseek-chat")
+
+# =====================================================================
+# 2. STRICT VALIDATION & INPUT SANITIZATION LAYER
+# =====================================================================
+
+DOMAIN_REGEX = re.compile(
+    r"^(?:[a-zA-Z0-9]"
+    r"(?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+"
+    r"[a-zA-Z]{2,63}$"
+)
+
+def validate_target_host(target: str) -> str:
+    """Validate whether input is a strict IPv4, IPv6, or FQDN string."""
+    target = target.strip()
+    try:
+        ipaddress.ip_address(target)
+        return target
+    except ValueError:
+        pass
+
+    if DOMAIN_REGEX.match(target):
+        return target
+
+    raise ValueError(f"Invalid target format: '{target}'. Must be a valid IP address or FQDN.")
+
+def validate_http_url(url: str) -> str:
+    """Validate HTTP/HTTPS URL scheme and network location."""
+    url = url.strip()
+    parsed = urllib.parse.urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise ValueError(f"Invalid URL scheme '{parsed.scheme}'. Only http and https are permitted.")
+    if not parsed.netloc:
+        raise ValueError("URL must contain a valid domain/host network location.")
+    return url
+
+
+class SubdomainEnumInput(BaseModel):
+    domain: str = Field(..., description="Target domain to inspect (e.g., example.com)")
+
+    @field_validator("domain")
+    @classmethod
+    def check_domain(cls, v: str) -> str:
+        return validate_target_host(v)
+
+
+class NmapScanInput(BaseModel):
+    target: str = Field(..., description="Target hostname or IP address to scan")
+
+    @field_validator("target")
+    @classmethod
+    def check_target(cls, v: str) -> str:
+        return validate_target_host(v)
+
+
+class WebScraperInput(BaseModel):
+    url: str = Field(..., description="Full HTTP/HTTPS URL to inspect")
+
+    @field_validator("url")
+    @classmethod
+    def check_url(cls, v: str) -> str:
+        return validate_http_url(v)
+
+
+class PacketCaptureInput(BaseModel):
+    pcap_path: str = Field(..., description="Path to local capture file (.pcap, .pcapng, .cap)")
+    display_filter: Optional[str] = Field(default="", description="Wireshark display filter string")
+
+    @field_validator("pcap_path")
+    @classmethod
+    def sanitize_path(cls, v: str) -> str:
+        clean = os.path.basename(v.strip())
+        if not clean.endswith((".pcap", ".pcapng", ".cap")):
+            raise ValueError("Target file must end with .pcap, .pcapng, or .cap")
+        return clean
+
+    @field_validator("display_filter")
+    @classmethod
+    def sanitize_filter(cls, v: Optional[str]) -> str:
+        if not v:
+            return ""
+        if not re.match(r"^[a-zA-Z0-9._\s=!<>&|()\"'-]+$", v):
+            raise ValueError("Display filter contains unsupported or disallowed characters.")
+        return v
+
+
+# =====================================================================
+# 3. ASYNCHRONOUS DIAGNOSTIC TOOL WRAPPERS
+# =====================================================================
+
+async def run_subdomain_enumerator(domain: str) -> Dict[str, Any]:
+    """Query certificate transparency logs (crt.sh) for domain asset mapping."""
+    try:
+        validated = SubdomainEnumInput(domain=domain)
+        target_domain = validated.domain
+        crt_url = f"https://crt.sh/?q=%25.{target_domain}&output=json"
+
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            response = await client.get(crt_url)
+            if response.status_code != 200:
+                return {
+                    "status": "error",
+                    "message": f"crt.sh query returned status code {response.status_code}",
+                    "subdomains": []
+                }
+            
+            data = response.json()
+            subdomains = set()
+            for entry in data:
+                name_value = entry.get("name_value", "")
+                for sub in name_value.split("\n"):
+                    sub = sub.strip().lower()
+                    if "*" not in sub and sub.endswith(target_domain):
+                        subdomains.add(sub)
+
+            return {
+                "status": "success",
+                "target": target_domain,
+                "count": len(subdomains),
+                "subdomains": sorted(list(subdomains))[:100]
+            }
+    except Exception as exc:
+        return {
+            "status": "error",
+            "message": f"Subdomain enumeration failed: {str(exc)}",
+            "subdomains": []
+        }
+
+
+async def run_nmap_port_scanner(target: str) -> Dict[str, Any]:
+    """Execute bounded service discovery on top 100 ports using subprocess execution."""
+    try:
+        validated = NmapScanInput(target=target)
+        target_host = validated.target
+
+        cmd = ["nmap", "-sV", "-T4", "--top-ports", "100", target_host]
+
+        process = await asyncio.create_subprocess_exec(
+            *cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE
+        )
+        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=45.0)
+
+        return {
+            "status": "success" if process.returncode == 0 else "warning",
+            "exit_code": process.returncode,
+            "stdout": stdout.decode("utf-8", errors="replace"),
+            "stderr": stderr.decode("utf-8", errors="replace")
+        }
+    except asyncio.TimeoutError:
+        return {
+            "status": "error",
+            "message": "Nmap scan execution timed out (45s limit)."
+        }
+    except Exception as exc:
+        return {
+            "status": "error",
+            "message": f"Nmap execution error: {str(exc)}"
+        }
+
+
+async def run_web_scraper_and_analyzer(url: str) -> Dict[str, Any]:
+    """Audit HTTP security response headers, meta tags, forms, and script sources."""
+    try:
+        validated = WebScraperInput(url=url)
+        target_url = validated.url
+
+        headers = {
+            "User-Agent": "Diagnostic-Audit-Agent/1.0 (Compliance; Non-Intrusive)"
+        }
+
+        async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client:
+            resp = await client.get(target_url, headers=headers)
+            soup = BeautifulSoup(resp.text, "html.parser")
+
+            sec_headers = {
+                "Strict-Transport-Security": resp.headers.get("strict-transport-security", "MISSING"),
+                "Content-Security-Policy": resp.headers.get("content-security-policy", "MISSING"),
+                "X-Frame-Options": resp.headers.get("x-frame-options", "MISSING"),
+                "X-Content-Type-Options": resp.headers.get("x-content-type-options", "MISSING"),
+                "Referrer-Policy": resp.headers.get("referrer-policy", "MISSING"),
+                "Server": resp.headers.get("server", "UNDISCLOSED")
+            }
+
+            forms = []
+            for f in soup.find_all("form")[:10]:
+                forms.append({
+                    "action": f.get("action", ""),
+                    "method": f.get("method", "GET").upper(),
+                    "inputs": [inp.get("name") for inp in f.find_all("input") if inp.get("name")]
+                })
+
+            scripts = [s.get("src") for s in soup.find_all("script") if s.get("src")][:15]
+
+            return {
+                "status": "success",
+                "effective_url": str(resp.url),
+                "http_status": resp.status_code,
+                "security_headers": sec_headers,
+                "cookies_detected": list(resp.cookies.keys()),
+                "forms_discovered": forms,
+                "external_scripts": scripts
+            }
+    except Exception as exc:
+        return {
+            "status": "error",
+            "message": f"Web analyzer error: {str(exc)}"
+        }
+
+
+async def run_packet_capture_analyzer(pcap_path: str, display_filter: str = "") -> Dict[str, Any]:
+    """Parse local packet captures using tshark with display filters."""
+    try:
+        validated = PacketCaptureInput(pcap_path=pcap_path, display_filter=display_filter)
+        target_path = validated.pcap_path
+        filt = validated.display_filter
+
+        if not os.path.exists(target_path):
+            return {
+                "status": "error",
+                "message": f"PCAP file '{target_path}' not found in runtime directory."
+            }
+
+        cmd = ["tshark", "-r", target_path, "-c", "50"]
+        if filt:
+            cmd.extend(["-Y", filt])
+
+        process = await asyncio.create_subprocess_exec(
+            *cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE
+        )
+        stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=20.0)
+
+        return {
+            "status": "success" if process.returncode == 0 else "warning",
+            "exit_code": process.returncode,
+            "packet_lines": stdout.decode("utf-8", errors="replace").splitlines()[:50],
+            "stderr": stderr.decode("utf-8", errors="replace")
+        }
+    except Exception as exc:
+        return {
+            "status": "error",
+            "message": f"Tshark analysis error: {str(exc)}"
+        }
+
+
+# =====================================================================
+# 4. TOOL REGISTRY & SCHEMA DEFINITIONS
+# =====================================================================
+
+DIAGNOSTIC_TOOLS = [
+    {
+        "type": "function",
+        "function": {
+            "name": "subdomain_enumerator",
+            "description": "Performs DNS asset discovery via certificate transparency records.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "domain": {
+                        "type": "string",
+                        "description": "Domain to audit (e.g. 'example.com')."
+                    }
+                },
+                "required": ["domain"]
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "nmap_port_scanner",
+            "description": "Performs service and port detection on the top 100 ports of a target host.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "target": {
+                        "type": "string",
+                        "description": "IPv4, IPv6, or FQDN to scan."
+                    }
+                },
+                "required": ["target"]
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "web_scraper_and_analyzer",
+            "description": "Extracts security response headers, forms, cookies, and linked assets from an HTTP/HTTPS endpoint.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "url": {
+                        "type": "string",
+                        "description": "Full HTTP or HTTPS URL to inspect."
+                    }
+                },
+                "required": ["url"]
+            }
+        }
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "packet_capture_analyzer",
+            "description": "Reads and filters network packet captures from a PCAP file using tshark.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "pcap_path": {
+                        "type": "string",
+                        "description": "Path to the PCAP file."
+                    },
+                    "display_filter": {
+                        "type": "string",
+                        "description": "Wireshark display filter (e.g. 'http or dns')."
+                    }
+                },
+                "required": ["pcap_path"]
+            }
+        }
+    }
+]
+
+TOOL_HANDLERS = {
+    "subdomain_enumerator": run_subdomain_enumerator,
+    "nmap_port_scanner": run_nmap_port_scanner,
+    "web_scraper_and_analyzer": run_web_scraper_and_analyzer,
+    "packet_capture_analyzer": run_packet_capture_analyzer
+}
+
+
+# =====================================================================
+# 5. RECURSIVE AGENTIC REACTION LOOP
+# =====================================================================
+
+SYSTEM_PROMPT = """You are an automated System Administration and Infrastructure Diagnostics Agent.
+You assist system administrators and network engineers in auditing network assets, inspecting security configurations, and mapping infrastructure surfaces.
+
+Operational Guidelines:
+1. Systematically investigate target infrastructure using the provided diagnostic tools.
+2. If tool outputs indicate further investigation is necessary (e.g., newly discovered subdomains or open services), call subsequent tools autonomously.
+3. Present final summaries structured with: Target Profile, Technical Diagnostics, Configuration Observations, and Remediation/Hardening Recommendations.
+4. Maintain a neutral, engineering-focused reporting format."""
+
+
+async def execute_tool(tool_name: str, arguments_json: str) -> str:
+    """Safely parse arguments and route to designated diagnostic wrapper."""
+    if tool_name not in TOOL_HANDLERS:
+        return json.dumps({"error": f"Tool '{tool_name}' not recognized."})
+
+    try:
+        kwargs = json.loads(arguments_json)
+    except json.JSONDecodeError:
+        return json.dumps({"error": f"Invalid JSON arguments supplied for '{tool_name}'."})
+
+    handler = TOOL_HANDLERS[tool_name]
+    try:
+        result = await handler(**kwargs)
+        return json.dumps(result, indent=2)
+    except Exception as exc:
+        return json.dumps({"error": f"Execution error in '{tool_name}': {str(exc)}"})
+
+
+async def run_agent_loop(
+    user_message: str,
+    chat_history: List[Dict[str, str]],
+    api_key: str,
+    base_url: str,
+    model_name: str,
+    max_steps: int = 5
+):
+    """
+    Executes the asynchronous multi-turn tool calling loop against the OpenAI-compatible API.
+    Streams execution progress, reasoning traces, and the final synthesis.
+    """
+    if not api_key:
+        yield chat_history + [{"role": "assistant", "content": "API Key is required to initialize the agent client."}], "Configuration Error: Missing API Key."
+        return
+
+    client = AsyncOpenAI(api_key=api_key, base_url=base_url)
+
+    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
+    for msg in chat_history:
+        messages.append({"role": msg["role"], "content": msg["content"]})
+    messages.append({"role": "user", "content": user_message})
+
+    logs = [f"**[INITIALIZE]** Audit Request: `{user_message}`\n"]
+    current_chat = chat_history + [{"role": "user", "content": user_message}]
+
+    step = 0
+    while step < max_steps:
+        step += 1
+        logs.append(f"\n**[ITERATION {step}/{max_steps}] Querying Model...**")
+        yield current_chat, "\n".join(logs)
+
+        try:
+            response = await client.chat.completions.create(
+                model=model_name,
+                messages=messages,
+                tools=DIAGNOSTIC_TOOLS,
+                tool_choice="auto",
+                temperature=0.2
+            )
+        except Exception as api_err:
+            err_text = f"API Request Failed: {str(api_err)}"
+            logs.append(f"❌ `{err_text}`")
+            current_chat.append({"role": "assistant", "content": err_text})
+            yield current_chat, "\n".join(logs)
+            return
+
+        choice = response.choices[0]
+        message = choice.message
+        reasoning = getattr(message, "reasoning_content", None)
+
+        if reasoning:
+            logs.append(f"\n🧠 **Analytical Reasoning:**\n```text\n{reasoning}\n```")
+            yield current_chat, "\n".join(logs)
+
+        messages.append(message)
+
+        if message.tool_calls:
+            for tool_call in message.tool_calls:
+                fn_name = tool_call.function.name
+                fn_args = tool_call.function.arguments
+
+                logs.append(f"⚙️ **Invoking:** `{fn_name}`\nArguments:\n```json\n{fn_args}\n```")
+                yield current_chat, "\n".join(logs)
+
+                tool_output = await execute_tool(fn_name, fn_args)
+
+                output_preview = tool_output if len(tool_output) <= 600 else f"{tool_output[:600]}... [truncated]"
+                logs.append(f"📊 **Result from `{fn_name}`:**\n```json\n{output_preview}\n```")
+                yield current_chat, "\n".join(logs)
+
+                messages.append({
+                    "role": "tool",
+                    "tool_call_id": tool_call.id,
+                    "name": fn_name,
+                    "content": tool_output
+                })
+        else:
+            final_response = message.content or "Diagnostic review complete."
+            logs.append("\n✅ **Audit Completed.**")
+            current_chat.append({"role": "assistant", "content": final_response})
+            yield current_chat, "\n".join(logs)
+            return
+
+    fallback_msg = "Diagnostic workflow exceeded maximum iteration depth."
+    current_chat.append({"role": "assistant", "content": fallback_msg})
+    yield current_chat, "\n".join(logs)
+
+
+# =====================================================================
+# 6. GRADIO USER INTERFACE
+# =====================================================================
+
+CUSTOM_CSS = """
+body, .gradio-container {
+    background-color: #0b0f19 !important;
+    color: #f1f5f9 !important;
+    font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
+}
+.gr-box, .gr-panel, .gr-form {
+    background-color: #111827 !important;
+    border-color: #1f2937 !important;
+}
+#terminal-output textarea {
+    background-color: #030712 !important;
+    color: #38bdf8 !important;
+    font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace !important;
+    font-size: 0.82rem !important;
+}
+"""
+
+def create_ui():
+    theme = gr.themes.Soft(
+        primary_hue="cyan",
+        secondary_hue="slate",
+        neutral_hue="slate"
+    )
+
+    with gr.Blocks(theme=theme, css=CUSTOM_CSS, title="Network Diagnostics Agent") as demo:
+        gr.Markdown(
+            """
+            # 🌐 Infrastructure Diagnostics & Asset Audit Platform
+            ### Asynchronous ReAct Diagnostic Engine with Containerized Tooling
+            """
+        )
+
+        with gr.Accordion("⚙️ Endpoint & API Settings", open=False):
+            with gr.Row():
+                api_key_input = gr.Textbox(
+                    label="API Key",
+                    type="password",
+                    value=DEFAULT_API_KEY,
+                    placeholder="Enter API Key"
+                )
+                base_url_input = gr.Textbox(
+                    label="Base URL",
+                    value=DEFAULT_BASE_URL
+                )
+                model_name_input = gr.Textbox(
+                    label="Model Identifier",
+                    value=DEFAULT_MODEL
+                )
+
+        with gr.Row():
+            with gr.Column(scale=5):
+                chatbot = gr.Chatbot(
+                    label="Diagnostic Workflow Session",
+                    height=520,
+                    type="messages",
+                    show_copy_button=True
+                )
+                with gr.Row():
+                    user_input = gr.Textbox(
+                        label="Audit Command / Target Specification",
+                        placeholder="E.g., Audit security headers and map open services for scanme.nmap.org",
+                        lines=2,
+                        scale=4
+                    )
+                    submit_btn = gr.Button("Run Audit", variant="primary", scale=1)
+
+                clear_btn = gr.ClearButton([user_input, chatbot], value="Clear Session")
+
+            with gr.Column(scale=4):
+                terminal_logs = gr.Markdown(
+                    value="*Execution logs, tool output traces, and model reasoning will stream here...*",
+                    elem_id="terminal-output"
+                )
+
+        async def on_submit(user_msg, chat_hist, key, url, model):
+            if not user_msg.strip():
+                yield chat_hist, "*No input provided.*"
+                return
+
+            chat_hist = chat_hist or []
+            async for updated_chat, updated_logs in run_agent_loop(
+                user_message=user_msg,
+                chat_history=chat_hist,
+                api_key=key,
+                base_url=url,
+                model_name=model
+            ):
+                yield updated_chat, updated_logs
+
+        submit_btn.click(
+            fn=on_submit,
+            inputs=[user_input, chatbot, api_key_input, base_url_input, model_name_input],
+            outputs=[chatbot, terminal_logs]
+        ).then(
+            fn=lambda: "",
+            inputs=None,
+            outputs=[user_input]
+        )
+
+    return demo
+
+
+if __name__ == "__main__":
+    app = create_ui()
+    app.queue(default_concurrency_limit=5).launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        show_api=False
+    )