| """CYPHER V12 M39 — Multimodal Chain-of-Thought. |
| |
| Extension of M10 multimodal_stub. For prompts containing images: |
| 1. Describe image via M10 (external API) |
| 2. Extract structured features (text/UI/diagram/screenshot) |
| 3. Build CoT prompt combining text + image features |
| 4. Generate response grounded in BOTH |
| |
| Cross-modal reasoning: e.g. "Analyze this network diagram + this PCAP" → |
| diagram describes infrastructure, PCAP describes traffic, response correlates. |
| """ |
| from __future__ import annotations |
|
|
| import logging |
| import re |
| from typing import Callable, Any |
|
|
| logger = logging.getLogger(__name__) |
|
|
|
|
| |
| _IMAGE_TYPE_PATTERNS = { |
| "screenshot": ("screenshot", "screen", "window", "dialog", "menu"), |
| "diagram": ("diagram", "architecture", "topology", "flowchart", "schema"), |
| "code": ("code", "snippet", "function", "class", "syntax highlight"), |
| "log": ("log", "console", "terminal", "output", "stack trace"), |
| "chart": ("chart", "graph", "axis", "plot", "candle", "ohlcv"), |
| "ui_form": ("form", "input field", "button", "login", "credential"), |
| "phishing": ("urgent", "verify", "suspended", "click here"), |
| } |
|
|
|
|
| def classify_image_type(description: str) -> list[str]: |
| """Detect image type from M10 description text.""" |
| d_lower = description.lower() |
| types: list[str] = [] |
| for itype, kws in _IMAGE_TYPE_PATTERNS.items(): |
| if any(kw in d_lower for kw in kws): |
| types.append(itype) |
| return types |
|
|
|
|
| def extract_image_facts(description: str) -> dict: |
| """Extract structured facts from image description.""" |
| facts: dict = {} |
| |
| ips = re.findall(r"\b(?:\d{1,3}\.){3}\d{1,3}\b", description) |
| urls = re.findall(r"https?://\S+", description) |
| domains = re.findall(r"\b[a-z0-9-]+\.(?:com|net|org|io|xyz|ru|cn)\b", description.lower()) |
| if ips: facts["ips"] = ips[:5] |
| if urls: facts["urls"] = urls[:5] |
| if domains: facts["domains"] = domains[:5] |
| |
| cves = re.findall(r"CVE-\d{4}-\d+", description, re.IGNORECASE) |
| if cves: facts["cves"] = list(set(c.upper() for c in cves))[:3] |
| |
| numbers = re.findall(r"\b\d{3,}\b", description) |
| if numbers: facts["large_numbers"] = numbers[:5] |
| |
| if "password" in description.lower() or "credentials" in description.lower(): |
| facts["credentials_signal"] = True |
| return facts |
|
|
|
|
| class MultimodalCoT: |
| """Cross-modal CoT: image + text → integrated reasoning.""" |
|
|
| def __init__( |
| self, |
| multimodal_stub=None, |
| generate_fn: Callable[[str, int, float], str] | None = None, |
| ): |
| self.mm = multimodal_stub |
| self.generate_fn = generate_fn |
|
|
| def analyze( |
| self, |
| prompt: str, |
| image_src: str | bytes | None = None, |
| text_context: str = "", |
| ) -> dict: |
| result: dict = {"prompt": prompt, "has_image": image_src is not None} |
|
|
| |
| image_description = "" |
| image_types: list[str] = [] |
| image_facts: dict = {} |
| if image_src is not None and self.mm: |
| try: |
| image_description = self.mm.describe_image( |
| image_src, |
| prompt="Describe this image for cybersecurity analysis. List visible text, UI elements, URLs, IPs, code, processes, or suspicious indicators." |
| ) |
| image_types = classify_image_type(image_description) |
| image_facts = extract_image_facts(image_description) |
| except Exception as e: |
| logger.warning(f"Image describe fail: {e}") |
| image_description = f"[Image description unavailable: {type(e).__name__}]" |
|
|
| result["image_description"] = image_description |
| result["image_types"] = image_types |
| result["image_facts"] = image_facts |
|
|
| |
| cot_parts: list[str] = [] |
| if image_description: |
| cot_parts.append(f"[IMAGE_DESCRIPTION: {image_description[:400]}]") |
| if image_types: |
| cot_parts.append(f"[IMAGE_TYPE: {','.join(image_types)}]") |
| if image_facts: |
| cot_parts.append(f"[IMAGE_FACTS: {image_facts}]") |
| if text_context: |
| cot_parts.append(f"[TEXT_CONTEXT: {text_context[:400]}]") |
| cot_parts.append(f"\nQuestion: {prompt}") |
| cot_parts.append("Analyze step by step (image observation → text correlation → conclusion):") |
| cot_prompt = "\n".join(cot_parts) |
| result["cot_prompt"] = cot_prompt |
|
|
| |
| if self.generate_fn: |
| try: |
| response = self.generate_fn(cot_prompt, 300, 0.35) |
| result["response"] = response |
| except Exception as e: |
| result["response"] = f"[Generation error: {e}]" |
| else: |
| result["response"] = "[no generate_fn provided]" |
|
|
| return result |
|
|
|
|
| __all__ = ["MultimodalCoT", "classify_image_type", "extract_image_facts"] |
|
|
|
|
| if __name__ == "__main__": |
| logging.basicConfig(level=logging.INFO) |
| print("=== M39 cypher_multimodal_cot SMOKE ===") |
| |
| class MockMM: |
| def describe_image(self, src, prompt=""): |
| return ("Screenshot of a Windows login form. URL https://micros0ft-security.com/login " |
| "shows fake Microsoft branding. User must enter password and verify. " |
| "Suspicious IP 185.220.101.5 in URL bar. Text says 'Your account is suspended, " |
| "click here urgent action required'.") |
|
|
| def mock_gen(prompt, mt, t): |
| return ("Step 1: Image shows phishing screenshot. " |
| "Step 2: Domain micros0ft-security.com is typosquat. " |
| "Step 3: IP 185.220.101.5 is Tor exit node. " |
| "Conclusion: Confirmed phishing site, do not enter credentials.") |
|
|
| mc = MultimodalCoT(multimodal_stub=MockMM(), generate_fn=mock_gen) |
| result = mc.analyze( |
| "Is this login page legitimate?", |
| image_src="/tmp/fake.png", |
| text_context="User clicked email link about account suspension.", |
| ) |
| print(f"\nImage types: {result['image_types']}") |
| print(f"Image facts: {result['image_facts']}") |
| print(f"CoT prompt (preview): {result['cot_prompt'][:300]}") |
| print(f"\nResponse: {result['response'][:300]}") |
| |
| print(f"\nclassify 'screenshot of dialog' → {classify_image_type('screenshot of a dialog box')}") |
| print(f"classify 'network diagram' → {classify_image_type('network architecture diagram')}") |
| print("=== SMOKE PASS ===") |
|
|
