RAVERSE 2.0 - AI Multi-Agent Binary Analysis & Patching System

📊 Package Distribution Status

NPM Package

PyPI Package

GitHub Repository

Project Status

Table of Contents

• Overview

• ⚠️ Legal Disclaimer & Responsible Use

• Key Features

• Getting Started: Real-World Usage Guide

  • Quick Start: Your First Analysis Workflow
  • Step-by-Step: Reverse Engineering Web Authentication
  • Understanding the Tool Chain
  • Cloud Infrastructure Overview
  • Common Workflows
  • Troubleshooting Tool Calls

• Technology Stack

• Quick Start

• System Architecture

• Database Architecture

• Agent Pipeline

• Agent Catalog

• Utilities Reference

• Configuration

• Memory & Knowledge Systems

• DeepCrawler API Discovery

• Monitoring & Metrics

• Performance & Scalability

• Usage Examples

• Deployment Workflows

  • Development Deployment
  • Production Deployment
  • Kubernetes Deployment
  • Cloudflare Workflows Deployment

• Development

• Documentation

• Contributing

• License

• Support

Overview

RAVERSE 2.0 is an advanced AI-powered multi-agent system for binary analysis, reverse engineering, and automated patching. It combines offline binary patching capabilities with online target analysis, leveraging multiple specialized AI agents to identify vulnerabilities, generate patches, and validate security improvements.

---

For free hosting of this project

---

⚠️ Legal Disclaimer & Responsible Use

CRITICAL: READ BEFORE USE

This section contains important legal information and usage restrictions. Failure to comply with these terms may result in criminal prosecution and civil liability.

1. Legal Disclaimer

RAVERSE 2.0 is provided "AS IS" without warranty of any kind, express or implied. The authors, maintainers, and contributors of RAVERSE 2.0 make no representations or warranties regarding the accuracy, completeness, or reliability of the software. Use of this software is entirely at your own risk.

The RAVERSE 2.0 project is designed for lawful, authorized security research and binary analysis only. Any use of this software for illegal purposes, unauthorized access, or malicious activities is strictly prohibited and will be prosecuted to the fullest extent of the law.

2. Liability Waiver

The authors, maintainers, and contributors of RAVERSE 2.0 are NOT liable for:

• Any damages, losses, or harm resulting from the use or misuse of this software
• Unauthorized access to systems, networks, or data
• Violation of computer fraud and abuse laws
• Breach of confidentiality or privacy
• Loss of data or system compromise
• Any criminal or civil penalties incurred by users
• Misuse of the software for malicious purposes
• Damages caused by third parties using this software

By using RAVERSE 2.0, you assume full responsibility for all consequences of your actions.

3. Authorization Requirements - MANDATORY

YOU MUST OBTAIN EXPLICIT WRITTEN AUTHORIZATION BEFORE:

Offline Binary Analysis

• ✋ Analyzing any binary, executable, or software you do not own or have explicit permission to analyze
• ✋ Reverse engineering proprietary software without written consent from the copyright holder
• ✋ Modifying or patching binaries belonging to third parties
• ✋ Extracting intellectual property or trade secrets from binaries

Online Target Analysis

• ✋ Conducting reconnaissance on any website, server, or infrastructure you do not own
• ✋ Performing network traffic interception (mitmproxy) on systems you do not control
• ✋ Discovering or testing APIs on third-party infrastructure
• ✋ Scanning for vulnerabilities on systems without explicit written permission
• ✋ Accessing or analyzing web applications you do not own
• ✋ Intercepting or analyzing network traffic from other users

Security Research

• ✋ Testing security vulnerabilities on production systems
• ✋ Conducting penetration testing without a signed contract
• ✋ Performing any form of security assessment on third-party infrastructure

"Written authorization" means a signed document from the system owner explicitly granting permission for the specific activities you intend to perform.

4. Ethical Use Guidelines

Users of RAVERSE 2.0 must adhere to the following ethical principles:

Responsible Security Research:

• Conduct security research only on systems you own or have explicit written permission to test
• Follow responsible disclosure practices when discovering vulnerabilities
• Report vulnerabilities to affected parties before public disclosure
• Allow reasonable time for vendors to patch before public disclosure (typically 90 days)
• Never exploit vulnerabilities for personal gain or malicious purposes

Responsible Disclosure:

• Notify affected organizations of discovered vulnerabilities through proper channels
• Provide sufficient technical details to enable remediation
• Avoid public disclosure until patches are available
• Respect embargo periods agreed upon with vendors
• Document all findings and communications

Ethical Boundaries:

• Respect privacy and confidentiality of all data encountered
• Do not access, modify, or exfiltrate data without authorization
• Do not use RAVERSE 2.0 to facilitate illegal activities
• Do not use RAVERSE 2.0 to harm individuals, organizations, or infrastructure
• Comply with all applicable laws and regulations in your jurisdiction

5. Prohibited Uses

The following uses of RAVERSE 2.0 are strictly prohibited:

• ❌ Unauthorized penetration testing or security assessments
• ❌ Unauthorized access to computer systems or networks (hacking)
• ❌ Malware creation, distribution, or analysis for malicious purposes
• ❌ Denial-of-service (DoS) or distributed denial-of-service (DDoS) attacks
• ❌ Unauthorized data exfiltration or theft
• ❌ Violation of the Computer Fraud and Abuse Act (CFAA) or equivalent laws
• ❌ Violation of the Digital Millennium Copyright Act (DMCA)
• ❌ Violation of GDPR, CCPA, or other data protection regulations
• ❌ Violation of intellectual property rights
• ❌ Violation of terms of service of any platform or service
• ❌ Circumventing security controls or authentication mechanisms
• ❌ Creating or distributing exploits for malicious purposes
• ❌ Facilitating cybercrime or criminal activity
• ❌ Violating privacy rights of individuals or organizations
• ❌ Any activity that could cause harm to individuals or infrastructure

6. Legal Consequences

Unauthorized use of RAVERSE 2.0 may result in:

Criminal Penalties:

• Computer Fraud and Abuse Act (CFAA) (United States): Up to 10 years imprisonment and $250,000 in fines for intentional unauthorized access
• Computer Misuse Act (United Kingdom): Up to 10 years imprisonment
• Criminal Code (Canada): Up to 10 years imprisonment
• Strafgesetzbuch (Germany): Up to 10 years imprisonment
• Cybercrime Laws (EU): Up to 5-10 years imprisonment depending on jurisdiction
• Local Cybersecurity Laws: Penalties vary by country and jurisdiction

Civil Penalties:

• Lawsuits for damages (potentially millions of dollars)
• Injunctions preventing further use of the software
• Restitution for damages caused
• Attorney fees and court costs

Professional Consequences:

• Loss of security clearances
• Termination of employment
• Permanent damage to professional reputation
• Exclusion from security research community
• Blacklisting by industry organizations

Regulatory Consequences:

• GDPR fines up to €20 million or 4% of annual revenue
• CCPA penalties up to $7,500 per violation
• Industry-specific regulatory penalties
• Compliance violations and sanctions

7. Compliance Checklist

Before using RAVERSE 2.0, verify:

• ✅ You own or have explicit written authorization for all systems you will analyze
• ✅ Your use complies with all applicable laws in your jurisdiction
• ✅ You have obtained written permission from system owners
• ✅ You understand the legal risks and consequences
• ✅ You will follow responsible disclosure practices
• ✅ You will not use RAVERSE 2.0 for illegal or malicious purposes
• ✅ You have informed your organization's legal team (if applicable)
• ✅ You have documented your authorization for audit purposes

8. Acknowledgment

By downloading, installing, or using RAVERSE 2.0, you acknowledge that:

1. You have read and understood this legal disclaimer
2. You accept full responsibility for your use of the software
3. You will use RAVERSE 2.0 only for lawful, authorized purposes
4. You will not hold the authors, maintainers, or contributors liable for any consequences
5. You understand the legal risks and potential criminal penalties
6. You will comply with all applicable laws and regulations
7. You will follow ethical security research practices
8. You will obtain written authorization before analyzing any systems you do not own

If you do not agree with these terms, do not use RAVERSE 2.0.

---

Key Features

• Multi-Agent Architecture: 21+ specialized AI agents for different analysis tasks
• Binary Patching Pipeline: Automated disassembly, analysis, patching, and verification
• Online Analysis: Remote target reconnaissance, traffic interception, API discovery
• Deep Research: Comprehensive web research and content analysis
• Memory Integration: Hierarchical and retrieval-based memory strategies
• Vector Search: Semantic similarity search using pgvector
• Production Ready: Docker containerization, monitoring, and deployment guides

Getting Started: Real-World Usage Guide

This is a practical, hands-on guide. Follow the steps exactly; paste the JSON blocks into your MCP client when invoking each tool. Never paste secrets into chat or files.

Security warning: Only analyze systems you own or have written permission to test. See the Legal Disclaimer above.

Quick Start: Your First Analysis Workflow

Goal: In ~5 minutes, discover auth behavior for your site and generate an OpenAPI sketch.

1. Set environment variables (PowerShell on Windows):

$env:BACKEND_URL="https://jaegis-raverse.onrender.com";
$env:PROXY_URL="https://raverse-mcp-proxy.use-manus-ai.workers.dev";
$env:DATABASE_URL="postgres://avnadmin:***@raverse-pg-db-raverse-pg-db.i.aivencloud.com:23055/defaultdb?sslmode=require";
$env:REDIS_URL="rediss://default:***@raverse-valkey-cache-raverse-pg-db.g.aivencloud.com:23056";
$env:POSTGRES_CA_CERT="-----BEGIN CERTIFICATE-----`n...your Aiven CA pem lines...`n-----END CERTIFICATE-----";
$env:LOG_LEVEL="INFO"; $env:SERVER_VERSION="1.0.11";

1. Start the server (no install):

npx -y raverse-mcp-server@latest

1. In your MCP client (e.g., Augment Code), select server "raverse", then run these tools in order, pasting the JSON:

• reconnaissance

{
  "target_url": "https://your-shop.example.com",
  "depth": 1,
  "include_js": true,
  "include_sitemaps": true
}

• fetch_content

{
  "url": "https://your-shop.example.com/_next/static/chunks/auth.abcdef.js",
  "timeout": 20,
  "retries": 1
}

• analyze_javascript

{
  "js_code": "<paste JS content here>",
  "deobfuscate": true,
  "signals": ["HMAC","JWT","nonce","timestamp","crypto","subtle"]
}

• traffic_interception (HAR mode)

{
  "mode": "har",
  "har": { "log": { "entries": [/* pasted HAR entries */] } }
}

• reverse_engineer_api

{
  "traffic_data": { "har": { "log": { "entries": [/* … */] } } },
  "js_analysis": { "functions": ["sign","getToken"], "algos": ["HMAC-SHA256"] },
  "emit_openapi": true
}

That’s it. You’ll get endpoints, auth clues, and an initial OpenAPI sketch.

Step-by-Step: Reverse Engineering Web Authentication

Use these eight tools to go from raw site → auth recipe → knowledge saved.

1. reconnaissance

{
  "target_url": "https://your-shop.example.com",
  "depth": 1,
  "include_js": true,
  "include_sitemaps": true
}

1. fetch_content

{
  "url": "https://your-shop.example.com/_next/static/chunks/auth.abcdef.js",
  "timeout": 20,
  "retries": 1
}

1. analyze_javascript

{
  "js_code": "<paste JS content here>",
  "deobfuscate": true,
  "signals": ["HMAC","JWT","nonce","timestamp","crypto","subtle"]
}

1. api_pattern_matcher

{
  "content": "…short snippet with fetch('/api/auth/token', {…})…",
  "hints": ["Authorization", "x-api-key", "x-signature", "bearer"]
}

1. traffic_interception (HAR mode)

{
  "mode": "har",
  "har": { "log": { "entries": [/* pasted HAR entries */] } }
}

1. reverse_engineer_api

{
  "traffic_data": { "har": { "log": { "entries": [/* … */] } } },
  "js_analysis": { "functions": ["sign","getToken"], "algos": ["HMAC-SHA256"] },
  "emit_openapi": true
}

1. response_classifier

{
  "samples": [{
    "url": "https://your-shop.example.com/api/orders",
    "status": 403,
    "headers": {"x-error-code":"sig_invalid"}
  }]
}

1. ingest_content (save your findings for RAG)

{
  "content": "Auth uses HMAC-SHA256(ts|method|path|body)…",
  "metadata": { "project":"shop", "type":"auth_notes", "env":"prod" }
}

Understanding the Tool Chain

• reconnaissance → discovers candidate JS + endpoints
• fetch_content → provides concrete JS for analysis
• analyze_javascript → emits functions/algos/signals used by auth
• traffic_interception → adds live request/response evidence (HAR)
• reverse_engineer_api → fuses JS + HAR into endpoints/auth/OpenAPI
• response_classifier → sanity-check responses and error codes
• ingest_content → persists knowledge for search_knowledge_base

Visual flow:

graph LR
    A[reconnaissance] --> B[fetch_content]
    B --> C[analyze_javascript]
    C --> D[traffic_interception]
    D --> E[reverse_engineer_api]

See the full tool reference in jaegis-RAVERSE-mcp-server/TOOLS_REGISTRY_COMPLETE.md.

Cloud Infrastructure Overview

• Render Backend API: https://jaegis-raverse.onrender.com
• Cloudflare Workers Proxy (MCP stdio isolation): https://raverse-mcp-proxy.use-manus-ai.workers.dev
• Aiven PostgreSQL (persistent state): postgres://avnadmin:***@raverse-pg-db-raverse-pg-db.i.aivencloud.com:23055/defaultdb?sslmode=require
• Aiven Valkey/Redis (cache, A2A messaging): rediss://default:***@raverse-valkey-cache-raverse-pg-db.g.aivencloud.com:23056

Set only the environment variables above—no code edits. The server auto-initializes DB tables on first write. Ingestion stores content in PostgreSQL; transient caches go to Redis.

Common Workflows

• Web analysis (auth focus): reconnaissance → fetch_content → analyze_javascript → traffic_interception → reverse_engineer_api → response_classifier → ingest_content
• Binary patching (offline): disassemble_binary → analyze_wasm (if applicable) → apply_patch → verify_patch
  • disassemble_binary

{ "binary_path": "./bin/target.exe", "architecture": "x86_64" }

• apply_patch

{ "binary_path": "./bin/target.exe", "patches": [{"offset":"0x401000","bytes":"90 90"}] }

• verify_patch

{ "binary_path": "./bin/target.exe", "checks": ["integrity","entrypoint"] }

• Knowledge base usage: search_knowledge_base after ingest_content

{ "query": "HMAC timestamp signature format", "top_k": 5 }

Binary + WASM Combined Analysis

Goal: Analyze a native binary that loads a WASM module, correlate behavior, patch both, and verify end-to-end.

1. disassemble_binary (native)

{
  "binary_path": "./bin/app_with_wasm.exe",
  "architecture": "x86_64"
}

1. analyze_wasm (module)

• If you have the WASM file on disk (e.g., ./bin/module.wasm), base64-encode its bytes and pass as wasm_data.

{
  "wasm_data": "AGFzbQEAAAABAAA..."
}

1. Correlate findings

• Map exported/imported function names from analyze_wasm to call sites and string refs in the native disassembly (e.g., functions like wasm_sign or sha256_update). Note intersections and control points.

1. apply_patch (native binary)

{
  "binary_path": "./bin/app_with_wasm.exe",
  "backup": true,
  "patches": [
    {"offset": "0x40123A", "bytes": "90 90"},
    {"offset": "0x402010", "bytes": "EB 0A"}
  ]
}

1. apply_patch (wasm module)

• Save the WASM module to disk (e.g., ./bin/module.wasm). Patch offsets refer to the raw module bytes.

{
  "binary_path": "./bin/module.wasm",
  "backup": true,
  "patches": [
    {"offset": "0x0000010", "bytes": "00"},
    {"offset": "0x00000A2", "bytes": "01"}
  ]
}

1. verify_patch (both components)

• Use the backup paths returned by apply_patch (or your own original copies).

{
  "original_binary": "./bin/app_with_wasm.exe.bak",
  "patched_binary": "./bin/app_with_wasm.exe"
}

{
  "original_binary": "./bin/module.wasm.bak",
  "patched_binary": "./bin/module.wasm"
}

Tips

• Keep small, testable patch deltas (first NOP a branch, then widen scope)
• Re-run wasm analysis after patches to confirm function sizes/indices stable
• Prefer patching the smallest component that achieves the goal (WASM vs native)

Other helpful tools you may use in this flow: api_pattern_matcher, response_classifier, analyze_wasm, database_query, cache_operation.

Troubleshooting Tool Calls

• Empty or noisy outputs: increase specificity (e.g., provide smaller JS slices to analyze_javascript; filter HAR to auth flows only)
• Timeouts: raise timeouts to 30–60s on fetch_content or provide retries=2
• HAR not parsed: ensure it’s a valid HTTP Archive (log.entries array present)
• Missing DB/Redis state: verify DATABASE_URL and REDIS_URL env vars; SSL required for Aiven
• Logs: set LOG_LEVEL=DEBUG; logs are written to stderr (never stdout) in MCP mode
• Cache collisions: change metadata.project or add a unique tag; Redis TTL will expire old cache automatically
• Connectivity: confirm Render and Aiven endpoints are reachable from your network
• MCP config and TLS guide: see docs/SECURITY_MCP_CONFIG.md

If a tool response looks off, copy the high-level fields (no secrets) and open an issue. We’ll help you correlate JS signals, HAR evidence, and endpoint auth quickly.

Technology Stack

Component	Technology
Language	Python 3.13+
Database	PostgreSQL 17 with pgvector
Cache	Redis 8.2
AI/LLM	OpenRouter API (Claude, GPT-4, Llama)
Binary Analysis	Capstone, pefile, pyelftools
Web Automation	Playwright, Selenium
Traffic Analysis	mitmproxy, scapy
Monitoring	Prometheus, Grafana
Containerization	Docker, Docker Compose

Quick Start

Prerequisites

• Python 3.13 or higher
• pip or poetry
• PostgreSQL 17 (for database features)
• Redis 8.2 (for caching)
• OpenRouter API key (for LLM features)

Installation

Option 1: NPX (Fastest - No Installation Required)

Run the MCP Server directly with NPX:

# Run the latest version without installation
npx raverse-mcp-server@latest

# Or with specific version
npx raverse-mcp-server@1.0.2

Option 2: Global NPM Installation

# Install globally
npm install -g raverse-mcp-server

# Run the server
raverse-mcp-server

Option 3: PyPI Installation

# Install from PyPI
pip install jaegis-raverse-mcp-server

# Run the server
python -m jaegis_raverse_mcp_server.server

Option 4: Clone Repository

1. Clone the repository:

git clone https://github.com/usemanusai/RAVERSE.git
cd RAVERSE

1. Create virtual environment:

python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

1. Install dependencies:

pip install -r requirements.txt

1. Configure environment:

cp .env.example .env
# Edit .env with your API keys and database credentials

1. Run the system:

# Offline binary analysis
python src/main.py path/to/binary.exe

# Online target analysis
python src/raverse_online_cli.py --scope examples/scope_example.json --options examples/options_example.json

System Architecture

High-Level Architecture

┌─────────────────────────────────────────────────────────────────┐
│                     RAVERSE 2.0 SYSTEM                          │
│                                                                 │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │         Orchestrator (Offline & Online)                  │   │
│  │  Coordinates agents, manages workflow, handles I/O       │   │
│  └──────────────────────────────────────────────────────────┘   │
│                            │                                    │
│          ┌─────────────────┼─────────────────┐                  │
│          │                 │                 │                  │
│     ┌────▼────┐      ┌─────▼─────┐     ┌─────▼─────┐            │
│     │Offline  │      │  Online   │     │ Advanced  │            │
│     │Pipeline │      │ Pipeline  │     │  Agents   │            │
│     │(DAA→    │      │ (Recon→   │     │ (RAG, KB, │            │
│     │LIMA→PEA │      │ Traffic→  │     │  Quality) │            │
│     │→VA)     │      │ JS→API→   │     │           │            │
│     │         │      │ WASM→Sec) │     │           │            │
│     └────┬────┘      └─────┬─────┘     └─────┬─────┘            │
│          │                 │                 │                  │
│          └─────────────────┼─────────────────┘                  │
│                            │                                    │
│          ┌─────────────────┼─────────────────┐                  │
│          │                 │                 │                  │
│    ┌─────▼────┐      ┌─────▼─────┐     ┌─────▼─────┐            │
│    │PostgreSQL│      │   Redis   │     │Prometheus │            │
│    │ +pgvector│      │   Cache   │     │  Metrics  │            │
│    │ (Persist)│      │(Fast I/O) │     │(Observ.)  │            │
│    └──────────┘      └───────────┘     └───────────┘            │
└─────────────────────────────────────────────────────────────────┘

Component Overview

• Orchestrator: Central coordinator managing agent lifecycle, workflow execution, and result aggregation
• Offline Pipeline: Binary analysis (DAA → LIMA → PEA → VA)
• Online Pipeline: Remote target analysis (8-phase reconnaissance to reporting)
• Advanced Agents: RAG, Knowledge Base, Quality Gate, Governance, Document Generation
• PostgreSQL: Persistent storage with pgvector for semantic search
• Redis: High-speed caching and agent-to-agent communication
• Prometheus: Metrics collection and monitoring

Data Flow

Offline Mode:

Binary Input → Disassembly Analysis (DAA) → Logic Identification (LIMA)
→ Patching Execution (PEA) → Verification (VA) → Patched Binary Output

Online Mode:

Target URL → Reconnaissance → Traffic Interception → JavaScript Analysis
→ API Reverse Engineering → WebAssembly Analysis → Security Analysis
→ Validation → Reporting

Project Structure

RAVERSE/
├── src/                          # Source code
│   ├── agents/                   # 21+ AI agent implementations
│   │   ├── orchestrator.py       # Main orchestration agent
│   │   ├── online_*.py           # Online analysis agents
│   │   └── ...
│   ├── utils/                    # Utility modules
│   │   ├── database.py           # PostgreSQL integration
│   │   ├── cache.py              # Redis caching
│   │   ├── embeddings.py         # Vector embeddings
│   │   └── ...
│   ├── config/                   # Configuration management
│   │   ├── settings.py           # Main settings
│   │   ├── agent_memory_config.py # Memory strategies
│   │   └── ...
│   ├── main.py                   # Offline analysis entry point
│   └── raverse_online_cli.py     # Online analysis CLI
├── tests/                        # Test suite (81+ tests)
│   ├── unit/                     # Unit tests
│   ├── integration/              # Integration tests
│   ├── deepcrawler/              # DeepCrawler tests
│   └── memory/                   # Memory integration tests
├── docs/                         # Documentation
│   ├── ARCHITECTURE.md           # System architecture
│   ├── PRODUCTION_DEPLOYMENT_GUIDE.md
│   ├── QUICK_START_AI_FEATURES.md
│   └── archive/                  # Historical documentation
├── examples/                     # Configuration examples
│   ├── scope_example.json        # Analysis scope config
│   ├── options_example.json      # Execution options
│   └── comprehensive_demo.py     # Demo script
├── scripts/                      # Automation scripts
│   ├── run_tests.ps1             # Test runner (PowerShell)
│   ├── run_tests.sh              # Test runner (Bash)
│   └── migrations/               # Database migrations
├── docker/                       # Docker infrastructure
│   ├── postgres/                 # PostgreSQL config
│   ├── redis/                    # Redis config
│   ├── prometheus/               # Prometheus config
│   └── grafana/                  # Grafana config
├── requirements.txt              # Python dependencies
├── Dockerfile                    # Container image
├── docker-compose.yml            # Multi-container setup
├── .env.example                  # Environment template
├── .gitignore                    # Git ignore rules
└── README.md                     # This file

Database Architecture

PostgreSQL with pgvector Integration

RAVERSE uses PostgreSQL 17 with the pgvector extension for semantic search capabilities, enabling vector similarity queries across code embeddings and analysis results.

Vector Search Implementation

• Embedding Dimensions:
  • Code embeddings: 384 dimensions (all-MiniLM-L6-v2 model)
  • Knowledge base: 1536 dimensions (OpenAI-compatible)
• Similarity Metrics: Cosine distance (<=> operator in pgvector)
• Indexing Strategy: HNSW (Hierarchical Navigable Small World) for O(log n) query performance
• Index Parameters: m=16, ef_construction=64 for balanced speed/accuracy

Core Tables

Table	Purpose	Key Columns
binaries	Binary file metadata	file_hash, file_type, architecture, status
disassembly_cache	Cached disassembly with embeddings	binary_id, address, instruction, embedding
code_embeddings	Code snippets with semantic vectors	binary_hash, code_snippet, embedding
vector_search_index	General semantic search index	content_type, content_id, embedding
knowledge_base	RAG knowledge store	content, embedding, source
rag_sessions	RAG query/response history	query, retrieved_knowledge, generated_response
logic_mappings	Control/data flow analysis	control_flow, data_flow, algorithms
analysis_results	Complete analysis outputs	binary_id, result_data, metadata

Vector Operations Example

-- Find similar instructions using cosine similarity
SELECT
    address, instruction, opcode,
    1 - (embedding <=> %s::vector) AS similarity
FROM disassembly_cache
WHERE 1 - (embedding <=> %s::vector) >= 0.7
ORDER BY embedding <=> %s::vector
LIMIT 10;

Performance Optimizations

• Connection Pooling: Maintains persistent connections to reduce overhead
• Batch Operations: Bulk inserts for embeddings and analysis results
• Query Caching: Redis caches frequent queries (TTL: 1 hour)
• Index Maintenance: Automatic VACUUM and ANALYZE on schedule

Redis Caching Layer

Redis 8.2 provides high-speed caching and agent-to-agent communication:

• LLM Response Cache: Caches OpenRouter API responses (TTL: 24 hours)
• Analysis Cache: Stores binary analysis results (TTL: 7 days)
• Embedding Cache: Caches generated embeddings (TTL: 7 days)
• A2A Communication: Redis Pub/Sub for agent message routing
• Session State: Temporary storage for agent execution state

Agent Architecture

Offline Agents (Binary Patching)

• DisassemblyAnalysisAgent: Analyzes binary disassembly
• LogicIdentificationMappingAgent: Maps code logic and flow
• PatchingExecutionAgent: Applies binary patches
• VerificationAgent: Validates patch integrity

Online Agents (Remote Analysis)

• ReconnaissanceAgent: Target reconnaissance
• TrafficInterceptionAgent: Network traffic analysis
• JavaScriptAnalysisAgent: JavaScript deobfuscation
• APIReverseEngineeringAgent: API discovery and analysis
• WebAssemblyAnalysisAgent: WASM analysis
• SecurityAnalysisAgent: Security vulnerability detection
• ValidationAgent: Result validation
• ReportingAgent: Report generation

Advanced Agents

• DeepResearchAgent: Comprehensive web research
• RAGOrchestratorAgent: Retrieval-augmented generation
• KnowledgeBaseAgent: Knowledge management
• VersionManagerAgent: Version tracking
• QualityGateAgent: Quality assurance

Agent Pipeline

Offline Binary Patching Pipeline

The offline pipeline executes sequentially through four core agents:

Binary File
    ↓
[DAA] Disassembly Analysis Agent
  • Extracts binary metadata (PE/ELF format, architecture)
  • Disassembles code using Capstone engine
  • Identifies functions and code sections
  • Generates instruction embeddings
    ↓
[LIMA] Logic Identification & Mapping Agent
  • Analyzes control flow (branches, loops, calls)
  • Analyzes data flow (register/memory operations)
  • Identifies algorithms and patterns
  • Generates flowcharts and logic maps
    ↓
[PEA] Patching Execution Agent
  • Converts virtual addresses to file offsets
  • Generates patch opcodes
  • Writes patches to binary
  • Creates backup before modification
    ↓
[VA] Verification Agent
  • Validates patch integrity
  • Verifies binary structure
  • Tests patched functionality
  • Generates verification report
    ↓
Patched Binary Output

Orchestration Logic (from src/agents/orchestrator.py):

• Binary metadata extraction and database recording
• Sequential agent execution with error handling
• Result caching in Redis and PostgreSQL
• Status tracking (processing → completed_success/completed_failed)
• Execution time monitoring and logging

Online Analysis Pipeline

The online pipeline executes 8 phases with parallel processing where applicable:

Target URL + Scope + Options
    ↓
[Phase 1] Reconnaissance Agent
  • Identifies technology stack
  • Discovers endpoints and services
  • Maps network topology
    ↓
[Phase 2] Traffic Interception Agent
  • Captures HTTP(S) traffic
  • Analyzes request/response patterns
  • Extracts API calls
    ↓
[Phase 3] JavaScript Analysis Agent
  • Deobfuscates JavaScript code
  • Extracts API calls from JS
  • Identifies client-side logic
    ↓
[Phase 4] API Reverse Engineering Agent
  • Maps API endpoints
  • Generates OpenAPI documentation
  • Identifies authentication methods
    ↓
[Phase 5] WebAssembly Analysis Agent
  • Decompiles WASM modules
  • Analyzes compiled code
  • Extracts functionality
    ↓
[Phase 6] Security Analysis Agent
  • Identifies vulnerabilities
  • Analyzes security headers
  • Checks for common weaknesses
    ↓
[Phase 7] Validation Agent
  • Generates proof-of-concept exploits
  • Validates findings
  • Captures evidence
    ↓
[Phase 8] Reporting Agent
  • Generates comprehensive reports
  • Formats findings (JSON, HTML, PDF)
  • Creates executive summary
    ↓
Analysis Report Output

Agent-to-Agent Communication (A2A Protocol)

Agents communicate via Redis Pub/Sub with PostgreSQL audit logging:

• Message Format: JSON with metadata (sender, receiver, correlation_id, priority)
• Channels: agent:messages:{receiver_agent} for routing
• Message Types: task_complete, data_request, error, status_update, ack
• Retry Logic: Exponential backoff with max 3 retries
• TTL: 3600 seconds (1 hour) for message expiration

Agent Catalog

RAVERSE 2.0 includes 35+ specialized AI agents organized into 5 categories:

Offline Binary Analysis Agents (4)

Agent	Purpose	Input	Output	Model
DisassemblyAnalysisAgent (DAA)	Binary disassembly & metadata extraction	Binary file path	Disassembly, functions, metadata	Capstone engine
LogicIdentificationMappingAgent (LIMA)	Control/data flow analysis	Disassembly output	Logic maps, flowcharts, algorithms	OpenRouter LLM
PatchingExecutionAgent (PEA)	Binary patching & modification	Logic maps + binary	Patched binary file	Binary utilities
VerificationAgent (VA)	Patch validation & integrity check	Patched binary	Verification report	Binary analysis

Online Analysis Agents (9)

Agent	Purpose	Input	Output	Model
ReconnaissanceAgent	Tech stack & endpoint discovery	Target URL	Tech stack, endpoints	Playwright + LLM
TrafficInterceptionAgent	HTTP(S) traffic capture & analysis	Target URL + duration	API calls, patterns	mitmproxy
JavaScriptAnalysisAgent	JS deobfuscation & analysis	JavaScript code	Deobfuscated code, APIs	OpenRouter LLM
APIReverseEngineeringAgent	API endpoint mapping	Traffic data + JS	OpenAPI spec, endpoints	OpenRouter LLM
WebAssemblyAnalysisAgent	WASM decompilation & analysis	WASM modules	Decompiled code, functions	Binary analysis
AICoPilotAgent	LLM-assisted analysis	Analysis context	Insights, recommendations	OpenRouter LLM
SecurityAnalysisAgent	Vulnerability detection	Analysis data	Vulnerabilities, risks	OpenRouter LLM
ValidationAgent	PoC generation & evidence capture	Findings	Validated findings, PoCs	Playwright + LLM
ReportingAgent	Multi-format report generation	All analysis data	Reports (JSON/HTML/PDF)	Document generation

Advanced Architecture Agents (8)

Agent	Purpose	Input	Output	Model
VersionManagerAgent	Version tracking & compatibility	Analysis data	Version info, compatibility	OpenRouter LLM
KnowledgeBaseAgent	Vector embeddings & RAG	Text content	Embeddings, knowledge store	Sentence-transformers
QualityGateAgent	Quality validation & metrics	Analysis results	Quality score, gate decision	OpenRouter LLM
GovernanceAgent	Strategic governance & approvals	Analysis data	Approval decision, governance	OpenRouter LLM
DocumentGeneratorAgent	Manifest & report generation	Analysis data	Documents, manifests	OpenRouter LLM
RAGOrchestratorAgent	Retrieval-augmented generation	Query + knowledge base	Generated response	OpenRouter LLM + pgvector
DeepResearchTopicEnhancerAgent	Research topic expansion	Research topic	Enhanced topics	OpenRouter LLM
DeepResearchWebResearcherAgent	Web research & content fetching	Research topics	Research findings	Playwright + LLM

Deep Research Agents (3)

Agent	Purpose	Input	Output	Model
DeepResearchContentAnalyzerAgent	Content analysis & synthesis	Web content	Analyzed content, insights	OpenRouter LLM
DeepResearchWebResearcherAgent	Comprehensive web research	Research queries	Research findings, sources	Playwright + LLM
DeepResearchTopicEnhancerAgent	Topic expansion & refinement	Research topics	Enhanced topics, subtopics	OpenRouter LLM

Utility & Support Agents (11+)

Agent	Purpose	Input	Output	Model
OnlineBaseAgent	Base class for online agents	Task data	Formatted result	Base implementation
OnlineOrchestrationAgent	Online pipeline orchestrator	Target URL + scope	Pipeline results	Orchestration logic
OrchestratingAgent	Offline pipeline orchestrator	Binary path	Analysis results	Orchestration logic
EnhancedOrchestratorAgent	Enhanced offline orchestrator	Binary path	Enhanced analysis	Orchestration logic
LLMAgent	Generic LLM interface	Prompt	LLM response	OpenRouter API
BaseMemoryAgent	Memory management base	Task data	Memory-augmented result	Memory strategies
A2AMixinAgent	Agent-to-agent communication	Message	Routed message	Redis Pub/Sub
APIPatternMatcherAgent	API pattern detection	Traffic data	Detected patterns	Pattern matching
DocumentGeneratorAgent	Document generation utility	Data	Generated documents	Document templates
ResponseClassifierAgent	Response classification	Response data	Classification	OpenRouter LLM
URLFrontierAgent	URL frontier management	URLs	Prioritized URLs	URL scheduling

Total Agents: 35+

Utilities Reference

RAVERSE includes 18+ utility modules providing core functionality:

Database & Persistence

database.py - PostgreSQL integration with connection pooling

• DatabaseManager: Main database interface
• create_binary_record(): Store binary metadata
• search_similar_instructions(): Vector similarity search
• execute_query(): Execute arbitrary SQL with retry logic
• Features: Connection pooling, transaction management, error handling

cache.py - Redis caching layer

• CacheManager: Redis interface
• cache_analysis(): Cache analysis results
• get_cached_llm_response(): Retrieve cached LLM responses
• Features: TTL management, key expiration, batch operations

Vector & Semantic Search

embeddings_v2.py - Embedding generation with caching

• EmbeddingGenerator: Generates embeddings using sentence-transformers
• generate_embedding(): Generate text embeddings (384-dim)
• generate_code_embedding(): Generate code-specific embeddings
• batch_encode(): Batch embedding generation with caching
• Features: Model caching, batch processing, metrics collection

semantic_search.py - Semantic code search engine

• SemanticSearchEngine: Vector similarity search
• store_code_embedding(): Store code with embeddings
• find_similar_code(): Find similar code snippets
• search_by_pattern(): Pattern-based search
• Features: Similarity thresholding, result ranking, metadata filtering

Binary Analysis

binary_utils.py - Binary file analysis utilities

• BinaryAnalyzer: PE/ELF binary analysis
• extract_metadata(): Extract binary metadata (format, arch, hash)
• va_to_offset(): Virtual address to file offset conversion
• get_sections(): Extract binary sections
• Features: Multi-format support (PE, ELF), architecture detection

Communication & Messaging

a2a_protocol.py - Agent-to-agent communication

• A2AProtocol: Redis Pub/Sub message routing
• publish_message(): Publish message to agent channel
• subscribe_to_channel(): Subscribe to agent messages
• format_message(): Format A2A protocol messages
• Features: Message validation, correlation tracking, audit logging

message_broker.py - Message brokering for agent coordination

• MessageBroker: Central message routing
• route_message(): Route messages between agents
• handle_response(): Process agent responses
• Features: Message queuing, priority handling, timeout management

Web & Content Fetching

content_fetcher.py - Web content retrieval

• ContentFetcher: HTTP(S) content fetching
• fetch_url(): Fetch webpage content
• extract_text(): Extract text from HTML
• Features: Retry logic, timeout handling, user-agent rotation

url_frontier.py - URL frontier management

• URLFrontier: Manages crawl frontier
• add_url(): Add URL to frontier
• get_next_url(): Get next URL to crawl
• mark_visited(): Mark URL as visited
• Features: Priority queue, duplicate detection, politeness delays

Analysis & Classification

api_pattern_matcher.py - API endpoint pattern detection

• APIPatternMatcher: Detects API patterns in traffic
• match_rest_api(): Identify REST API patterns
• match_graphql(): Identify GraphQL patterns
• extract_endpoints(): Extract API endpoints
• Features: Pattern library, confidence scoring, metadata extraction

response_classifier.py - Response type classification

• ResponseClassifier: Classifies HTTP responses
• classify_response(): Determine response type
• extract_schema(): Extract response schema
• Features: Content-type detection, schema inference

websocket_analyzer.py - WebSocket protocol analysis

• WebSocketAnalyzer: Analyzes WebSocket connections
• analyze_handshake(): Analyze WS handshake
• extract_messages(): Extract WS messages
• Features: Protocol version detection, message parsing

Scheduling & Crawling

crawl_scheduler.py - Crawl scheduling and coordination

• CrawlScheduler: Manages crawl scheduling
• schedule_crawl(): Schedule crawl job
• get_next_job(): Get next scheduled job
• Features: Priority scheduling, rate limiting, job persistence

Metrics & Monitoring

metrics.py - Prometheus metrics collection

• MetricsCollector: Collects system metrics
• record_agent_execution(): Record agent execution time
• record_embedding_generation(): Record embedding metrics
• record_cache_hit(): Record cache statistics
• Features: Prometheus export, metric aggregation, time-series data

Multi-Level Caching

multi_level_cache.py - Hierarchical caching strategy

• MultiLevelCache: L1 (memory) + L2 (Redis) + L3 (PostgreSQL)
• get(): Retrieve from cache hierarchy
• set(): Store in cache hierarchy
• invalidate(): Invalidate cache entries
• Features: Automatic promotion, TTL management, consistency

Usage Examples

Binary Analysis - Offline Pipeline

Basic Usage

from src.agents.orchestrator import OrchestratingAgent

# Initialize orchestrator with OpenRouter API
oa = OrchestratingAgent(
    openrouter_api_key="sk-or-v1-your-key",
    model="meta-llama/llama-3.3-70b-instruct:free",
    use_database=True
)

# Analyze binary file
result = oa.run("path/to/binary.exe")

# Result structure
print(f"Success: {result.get('success')}")
print(f"Patches Applied: {result.get('patches_applied')}")
print(f"Verification: {result.get('verification_status')}")

Advanced Usage with Database

from src.agents.orchestrator import OrchestratingAgent
from src.utils.database import DatabaseManager
from src.utils.cache import CacheManager

# Initialize with database and cache
oa = OrchestratingAgent(use_database=True)

# Analyze binary
result = oa.run("path/to/binary.exe")

# Query analysis results from database
db = DatabaseManager()
binary_records = db.execute_query(
    "SELECT * FROM raverse.binaries WHERE file_hash = %s",
    (result['binary_hash'],)
)

# Retrieve cached results
cache = CacheManager()
cached_result = cache.get_cached_analysis(result['binary_hash'])

Semantic Code Search

from src.utils.semantic_search import SemanticSearchEngine
from src.utils.database import DatabaseManager
from src.utils.cache import CacheManager

# Initialize search engine
db = DatabaseManager()
cache = CacheManager()
search_engine = SemanticSearchEngine(db, cache)

# Store code snippet with embedding
search_engine.store_code_embedding(
    binary_hash="abc123def456",
    code_snippet="cmp eax, 0x0; je 0x401000",
    metadata={"function": "main", "offset": "0x401000"}
)

# Find similar code
results = search_engine.find_similar_code(
    query="compare eax with zero and jump if equal",
    limit=10,
    similarity_threshold=0.7
)

for result in results:
    print(f"Similarity: {result['similarity']:.2%}")
    print(f"Code: {result['code_snippet']}")
    print(f"Binary: {result['binary_hash'][:8]}...")

Online Analysis - Remote Target Analysis

Basic Online Analysis

python src/raverse_online_cli.py \
  --target https://api.example.com \
  --scope examples/scope_example.json \
  --options examples/options_example.json \
  --output results/

Scope Configuration (scope_example.json)

{
  "target_url": "https://api.example.com",
  "allowed_domains": ["api.example.com", "*.example.com"],
  "excluded_paths": ["/admin", "/internal"],
  "max_depth": 3,
  "max_urls": 1000
}

Options Configuration (options_example.json)

{
  "recon": {
    "detect_technologies": true,
    "detect_endpoints": true
  },
  "traffic": {
    "duration_seconds": 60,
    "capture_ssl": true
  },
  "api_discovery": {
    "detect_rest": true,
    "detect_graphql": true,
    "detect_websockets": true
  },
  "security": {
    "check_vulnerabilities": true,
    "generate_poc": true
  }
}

Programmatic Online Analysis

from src.agents.online_orchestrator import OnlineOrchestrationAgent

# Initialize online orchestrator
oa = OnlineOrchestrationAgent(
    api_key="sk-or-v1-your-key",
    model="meta-llama/llama-3.3-70b-instruct:free"
)

# Execute online analysis
result = oa.execute(
    target_url="https://api.example.com",
    scope={
        "target_url": "https://api.example.com",
        "allowed_domains": ["api.example.com"],
        "max_depth": 3
    },
    options={
        "recon": {"detect_technologies": True},
        "traffic": {"duration_seconds": 60},
        "api_discovery": {"detect_rest": True}
    }
)

# Access results
print(f"Reconnaissance: {result['recon']}")
print(f"APIs Discovered: {result['api_reeng']}")
print(f"Vulnerabilities: {result['security']}")

RAG (Retrieval-Augmented Generation) Usage

from src.agents.online_rag_orchestrator_agent import RAGOrchestratorAgent
from src.utils.semantic_search import SemanticSearchEngine

# Initialize RAG orchestrator
rag = RAGOrchestratorAgent(
    api_key="sk-or-v1-your-key",
    model="meta-llama/llama-3.3-70b-instruct:free"
)

# Execute RAG query
result = rag.execute({
    "query": "What are common binary patching techniques?",
    "context": "Binary analysis and security patching"
})

# Result includes retrieved knowledge + generated response
print(f"Retrieved Knowledge: {result['retrieved_knowledge']}")
print(f"Generated Response: {result['generated_response']}")
print(f"Confidence: {result['confidence']}")

Memory Configuration Usage

from src.config.agent_memory_config import AGENT_MEMORY_CONFIG, MEMORY_PRESETS

# Get memory configuration for specific agent
kb_config = AGENT_MEMORY_CONFIG['knowledge_base']
print(f"Strategy: {kb_config['strategy']}")
print(f"Preset: {kb_config['preset']}")
print(f"Reason: {kb_config['reason']}")

# Get preset details
heavy_preset = MEMORY_PRESETS['heavy']
print(f"Description: {heavy_preset['description']}")
print(f"RAM: {heavy_preset['ram_mb']} MB")
print(f"CPU: {heavy_preset['cpu_percent']}%")

Running Tests

All Tests

pytest tests/ -v --cov=src --cov-report=html

Specific Test Suites

# Unit tests
pytest tests/unit/ -v

# Integration tests
pytest tests/integration/ -v

# DeepCrawler tests
pytest tests/deepcrawler/ -v

# Memory integration tests
pytest tests/memory/ -v

# Complete architecture tests
pytest tests/test_complete_architecture.py -v

Test with Markers

# Run only fast tests
pytest tests/ -m "not slow" -v

# Run only integration tests
pytest tests/ -m "integration" -v

# Run with specific keyword
pytest tests/ -k "orchestrator" -v

Coverage Report

# Generate HTML coverage report
pytest tests/ --cov=src --cov-report=html

# View report
open htmlcov/index.html

Configuration

Configuration Files

File	Purpose	Location
settings.py	Main application settings	src/config/
agent_memory_config.py	Agent memory strategies	src/config/
deepcrawler_config.py	DeepCrawler parameters	src/config/
binary_analysis_settings.py	Binary analysis options	src/config/
deep_research_settings.py	Deep research configuration	src/config/
knowledge_base_settings.py	Knowledge base setup	src/config/
governance_settings.py	Governance rules	src/config/
quality_gate_settings.py	Quality gate thresholds	src/config/

Environment Variables

All settings can be configured via environment variables in .env:

API Configuration

OPENROUTER_API_KEY=sk-or-v1-your-key-here
OPENROUTER_MODEL=meta-llama/llama-3.3-70b-instruct:free

Database Configuration

DB_HOST=localhost
DB_PORT=5432
DB_USER=raverse
DB_PASSWORD=your_password
DB_NAME=raverse_db

Redis Configuration

REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_DB=0

Logging Configuration

LOG_LEVEL=INFO
LOG_FILE=logs/raverse.log

DeepCrawler Configuration

DEEPCRAWLER_MAX_DEPTH=3
DEEPCRAWLER_MAX_URLS=10000
DEEPCRAWLER_MAX_CONCURRENT=5
DEEPCRAWLER_TIMEOUT=30
DEEPCRAWLER_RATE_LIMIT=20.0

Memory Configuration

MEMORY_PRESET=medium  # none, light, medium, heavy

Configuration Precedence

1. Environment Variables (highest priority)
2. Configuration Files (src/config/*.py)
3. Default Values (lowest priority)

Memory Presets

Preset	Strategy	RAM	CPU	Use Case
none	No memory	0 MB	0%	Default, zero overhead
light	Sliding window	5 MB	1%	Short conversations
medium	Hierarchical	20 MB	3%	Balanced approach
heavy	Retrieval/RAG	100 MB	5%	Long conversations, semantic search

Agent-Specific Memory Configuration

Each agent has recommended memory strategy (from agent_memory_config.py):

• VersionManager: Hierarchical (medium) - Critical version info retention
• KnowledgeBase: Retrieval (heavy) - Semantic search for knowledge
• QualityGate: Memory-Augmented (medium) - Critical metrics + context
• Governance: Hierarchical (medium) - Approval rules + history
• DocumentGenerator: Summarization (medium) - Long documents + token efficiency
• RAGOrchestrator: Retrieval (heavy) - Semantic search + knowledge relationships
• DAA/LIMA: OS-Like (heavy) - Large binaries + virtual memory
• Online Agents: Sliding Window (light) - Minimal memory overhead

See .env.example for all available options.

Memory & Knowledge Systems

Memory Strategies

RAVERSE implements multiple memory strategies for different agent requirements:

1. Hierarchical Memory

• Use Case: Version management, governance, quality gates
• Structure: Multi-level hierarchy (recent → important → archived)
• Window Size: 3 messages (configurable)
• Retention: Long-term critical information
• Example: Version compatibility tracking across analysis runs

2. Retrieval-Based Memory (RAG)

• Use Case: Knowledge base, RAG orchestrator
• Mechanism: Vector similarity search using pgvector
• Embedding Dimensions: 384-1536 (model-dependent)
• Similarity Metric: Cosine distance
• Retrieval: Top-k results with threshold filtering
• Example: Finding similar code patterns across binaries

3. Memory-Augmented

• Use Case: Quality gates, validation agents
• Combination: Hierarchical + retrieval strategies
• Window Size: 2 messages + semantic search
• Retention: Recent context + relevant historical data
• Example: Quality metrics + historical thresholds

4. Sliding Window

• Use Case: Online agents, reconnaissance
• Window Size: 2-3 messages (minimal overhead)
• Retention: Only recent context
• Memory: ~5 MB per agent
• Example: Traffic interception agent tracking recent requests

5. OS-Like Memory

• Use Case: Binary analysis agents (DAA, LIMA)
• Structure: Virtual memory simulation
• RAM Size: 3 segments (configurable)
• Paging: Automatic overflow to disk
• Retention: Large binary analysis state
• Example: Handling multi-GB binary files

6. Summarization

• Use Case: Document generation, reporting
• Mechanism: Automatic context summarization
• Threshold: 4 messages before summarization
• Token Efficiency: Reduces context window usage
• Example: Summarizing long analysis reports

RAG (Retrieval-Augmented Generation) Architecture

RAG enhances LLM responses by retrieving relevant knowledge before generation:

Query Input
    ↓
[Embedding Generation]
  Generate query embedding (384-dim)
    ↓
[Vector Similarity Search]
  Search knowledge_base table using pgvector
  Cosine similarity with threshold (0.7)
    ↓
[Retrieved Context]
  Top-k results (k=5 default)
  Ranked by similarity score
    ↓
[Prompt Augmentation]
  Combine query + retrieved context
  Maintain token budget
    ↓
[LLM Generation]
  OpenRouter API call
  Generate response with context
    ↓
[Response Output]
  Formatted result with sources

Knowledge Base Management

Storage: PostgreSQL knowledge_base table

• Columns: knowledge_id, content, embedding, metadata, source, created_at
• Indexing: HNSW index on embedding column
• Capacity: Unlimited (scales with PostgreSQL)

Embedding Generation:

• Model: all-MiniLM-L6-v2 (384 dimensions)
• Batch Size: 32 (configurable)
• Caching: Redis cache (TTL: 7 days)
• Performance: ~100 embeddings/second

Retrieval Process:

-- Find top-k similar knowledge
SELECT
    knowledge_id, content, metadata,
    1 - (embedding <=> query_embedding::vector) AS similarity
FROM knowledge_base
WHERE 1 - (embedding <=> query_embedding::vector) >= 0.7
ORDER BY embedding <=> query_embedding::vector
LIMIT 5;

Context Management

• Context Window: 4096 tokens (configurable per model)
• Token Budget: 70% for context, 30% for generation
• Pruning: Automatic removal of low-relevance context
• Compression: Summarization for long contexts

Vector Similarity Configuration

• Similarity Metric: Cosine distance (1 - dot product)
• Threshold: 0.7 (70% similarity minimum)
• Top-K: 5 results by default
• Ranking: By similarity score (descending)

DeepCrawler API Discovery

Purpose

DeepCrawler is an automated API discovery and documentation system that:

• Crawls web applications to discover API endpoints
• Intercepts traffic to identify API calls
• Generates OpenAPI/Swagger documentation
• Detects REST, GraphQL, and WebSocket APIs
• Extracts authentication requirements

Architecture

Target Application
    ↓
[Browser Automation] (Playwright)
  • Headless browser navigation
  • JavaScript execution
  • Form interaction
    ↓
[Traffic Interception] (mitmproxy)
  • HTTP(S) traffic capture
  • Request/response analysis
  • API call extraction
    ↓
[API Pattern Detection]
  • REST endpoint identification
  • GraphQL query detection
  • WebSocket connection tracking
    ↓
[Endpoint Analysis]
  • HTTP method detection
  • Parameter extraction
  • Authentication analysis
    ↓
[Documentation Generation]
  • OpenAPI spec creation
  • Endpoint cataloging
  • Example generation
    ↓
API Documentation Output

Configuration

From src/config/deepcrawler_config.py:

# Crawling parameters
max_depth: int = 3              # Maximum crawl depth
max_urls: int = 10000           # Maximum URLs to crawl
max_concurrent: int = 5         # Concurrent requests
timeout: int = 30               # Request timeout (seconds)
rate_limit: float = 20.0        # Requests per minute

# API detection
detect_rest_apis: bool = True
detect_graphql: bool = True
detect_websockets: bool = True
min_confidence_score: float = 0.6

# Output
output_format: str = 'openapi'  # openapi, json, yaml
output_dir: str = './crawl_results'

Usage Example

python src/raverse_online_cli.py \
  --target https://api.example.com \
  --scope examples/scope_example.json \
  --options examples/options_example.json \
  --output results/

Output Formats

OpenAPI 3.0 Specification:

{
  "openapi": "3.0.0",
  "info": {
    "title": "Discovered API",
    "version": "1.0.0"
  },
  "paths": {
    "/api/users": {
      "get": {
        "summary": "List users",
        "parameters": [...],
        "responses": {...}
      }
    }
  }
}

JSON Format:

{
  "endpoints": [
    {
      "url": "/api/users",
      "method": "GET",
      "parameters": [...],
      "authentication": "Bearer token",
      "confidence": 0.95
    }
  ]
}

Database Schema

DeepCrawler stores results in PostgreSQL:

Table	Purpose
crawl_sessions	Crawl job metadata
discovered_apis	Discovered API endpoints
api_parameters	Endpoint parameters
api_authentication	Authentication methods
crawl_results	Raw crawl data

Docker Deployment

# Build and run with Docker Compose
docker-compose up -d

# View logs
docker-compose logs -f

# Stop services
docker-compose down

Monitoring & Metrics

Prometheus Metrics

RAVERSE exposes Prometheus metrics for monitoring:

Agent Execution Metrics

• agent_execution_duration_seconds: Time to execute agent (histogram)
• agent_execution_total: Total agent executions (counter)
• agent_execution_errors_total: Failed agent executions (counter)
• agent_state: Current agent state (gauge)

Database Metrics

• database_query_duration_seconds: Query execution time (histogram)
• database_connection_pool_size: Active connections (gauge)
• database_query_errors_total: Failed queries (counter)

Cache Metrics

• cache_hit_ratio: Cache hit rate (gauge)
• cache_operations_total: Total cache operations (counter)
• cache_evictions_total: Cache evictions (counter)

Embedding Metrics

• embedding_generation_duration_seconds: Embedding generation time (histogram)
• embedding_cache_hit_ratio: Embedding cache hit rate (gauge)
• embeddings_generated_total: Total embeddings generated (counter)

Vector Search Metrics

• vector_search_duration_seconds: Search query time (histogram)
• vector_search_results_count: Results per query (histogram)
• vector_search_similarity_score: Similarity scores (histogram)

Grafana Dashboards

Available dashboards in docker/grafana/:

1. System Overview: CPU, memory, disk, network
2. Agent Performance: Execution times, success rates, error rates
3. Database Metrics: Query performance, connection pool, cache efficiency
4. Vector Search: Search latency, result quality, index performance
5. API Discovery: Crawl progress, endpoints discovered, confidence scores

Key Performance Indicators (KPIs)

KPI	Target	Measurement
Binary Analysis Success Rate	>95%	Successful analyses / total
Average Analysis Time	<5s	Mean execution time
API Discovery Accuracy	>90%	Correctly identified endpoints / total
Cache Hit Ratio	>70%	Cache hits / total requests
Vector Search Latency	<100ms	p95 query time
Agent Error Rate	<5%	Failed executions / total

Logging Strategy

Log Levels

• DEBUG: Detailed execution flow, variable values
• INFO: Major milestones, agent transitions
• WARNING: Recoverable errors, fallbacks
• ERROR: Unrecoverable errors, failures
• CRITICAL: System-level failures

Log Destinations

• File: logs/raverse.log (rotating, 100MB per file)
• Stdout: Console output for container environments
• Structured Logging: JSON format for log aggregation

Structured Log Format

{
  "timestamp": "2025-10-26T10:30:00Z",
  "level": "INFO",
  "logger": "orchestrator",
  "message": "Starting DAA.disassemble",
  "binary_id": "abc123",
  "execution_time_ms": 1234,
  "tags": ["offline", "analysis"]
}

Accessing Monitoring

Prometheus UI:

http://localhost:9090

Grafana UI:

http://localhost:3000
Default credentials: admin / admin

Logs:

# View live logs
docker-compose logs -f raverse

# View specific service logs
docker-compose logs -f postgres
docker-compose logs -f redis

Performance & Scalability

Benchmarks

Binary Analysis Performance

• Small Binary (<1 MB): ~2-3 seconds
• Medium Binary (1-10 MB): ~5-10 seconds
• Large Binary (10-100 MB): ~30-60 seconds
• Bottleneck: Disassembly (Capstone) and LLM analysis

API Discovery Performance

• Simple Target (10-50 endpoints): ~30-60 seconds
• Complex Target (50-200 endpoints): ~2-5 minutes
• Large Target (200+ endpoints): ~10-30 minutes
• Bottleneck: Traffic interception and pattern matching

Vector Search Performance

• Query Latency: <100ms (p95) with HNSW index
• Throughput: 1000+ queries/second
• Index Size: ~1 MB per 10,000 embeddings

Resource Requirements

Minimum Configuration

• CPU: 2 cores
• RAM: 4 GB
• Disk: 20 GB
• Network: 10 Mbps

Recommended Configuration

• CPU: 8 cores
• RAM: 16 GB
• Disk: 100 GB SSD
• Network: 100 Mbps

Per-Component Breakdown

Component	CPU	RAM	Disk
Python Agents	2-4 cores	2-4 GB	1 GB
PostgreSQL	2 cores	4-8 GB	50-100 GB
Redis	1 core	2-4 GB	10 GB
Prometheus	1 core	1-2 GB	10 GB
Grafana	1 core	512 MB	1 GB

Scalability Limits

• Concurrent Analyses: 10-20 (limited by LLM API rate limits)
• Database Size: Unlimited (scales with PostgreSQL)
• Cache Size: Limited by Redis memory (default 2 GB)
• Vector Index: 1M+ embeddings (with proper indexing)

Optimization Techniques

Connection Pooling

# PostgreSQL connection pooling
pool_size = 10
max_overflow = 20
pool_recycle = 3600

Batch Processing

# Batch embedding generation
batch_size = 32
# Batch database inserts
insert_batch_size = 1000

Caching Strategy

• L1 Cache: In-memory (Python dict)
• L2 Cache: Redis (fast, distributed)
• L3 Cache: PostgreSQL (persistent)

Async Operations

• Agent execution: Parallel where possible
• I/O operations: Non-blocking
• LLM calls: Concurrent requests with rate limiting

Scaling Strategies

Horizontal Scaling

• Multiple agent worker processes
• Load balancing across workers
• Distributed Redis cluster
• PostgreSQL read replicas

Vertical Scaling

• Increase CPU cores for LLM inference
• Increase RAM for caching
• SSD storage for database
• Network bandwidth for concurrent operations

Documentation

• Architecture Guide
• Production Deployment
• Quick Start - AI Features
• DeepCrawler User Guide
• Memory Integration Guide

Advanced Configuration

Database Connection Pooling

Configure PostgreSQL connection pooling for production:

from src.utils.database import DatabaseManager

# Initialize with custom pool settings
db = DatabaseManager(
    pool_size=10,           # Minimum connections
    max_overflow=20,        # Maximum overflow connections
    pool_recycle=3600,      # Recycle connections after 1 hour
    pool_pre_ping=True      # Test connections before use
)

Redis Cluster Configuration

For distributed caching:

from src.utils.cache import CacheManager

# Initialize with Redis cluster
cache = CacheManager(
    redis_nodes=[
        ('redis-node-1', 6379),
        ('redis-node-2', 6379),
        ('redis-node-3', 6379)
    ],
    cluster_mode=True,
    skip_full_coverage_check=True
)

LLM Model Selection

Available models via OpenRouter:

# Free models (rate-limited)
models = [
    "meta-llama/llama-3.3-70b-instruct:free",
    "meta-llama/llama-3.2-3b-instruct:free",
    "mistralai/mistral-7b-instruct:free"
]

# Premium models (faster, better quality)
premium_models = [
    "openai/gpt-4-turbo",
    "anthropic/claude-3-opus",
    "google/gemini-pro"
]

# Select model based on use case
from src.agents.orchestrator import OrchestratingAgent

oa = OrchestratingAgent(model="openai/gpt-4-turbo")

Embedding Model Configuration

from src.utils.embeddings_v2 import EmbeddingGenerator

# Initialize with specific model
embedding_gen = EmbeddingGenerator(
    model_name="all-MiniLM-L6-v2",  # 384 dimensions
    batch_size=32,
    cache_manager=cache_manager
)

# Generate embeddings
embeddings = embedding_gen.batch_encode(
    texts=["code snippet 1", "code snippet 2"],
    show_progress_bar=True
)

Custom Agent Implementation

Create custom agents by extending base classes:

from src.agents.online_base_agent import OnlineBaseAgent

class CustomAnalysisAgent(OnlineBaseAgent):
    """Custom agent for specialized analysis."""

    def __init__(self, orchestrator, api_key, model):
        super().__init__(
            name="CustomAnalysis",
            orchestrator=orchestrator,
            api_key=api_key,
            model=model
        )

    def _execute_impl(self, task):
        """Implement custom analysis logic."""
        # Your implementation here
        return {
            "status": "success",
            "results": {...}
        }

Performance Tuning

Optimize Vector Search

# Adjust HNSW index parameters
# In PostgreSQL:
CREATE INDEX idx_embeddings_hnsw ON embeddings
USING hnsw (embedding vector_cosine_ops)
WITH (m = 32, ef_construction = 128);  # Higher values = better quality, slower

# Query optimization
SET hnsw.ef_search = 200;  # Higher = more accurate, slower

Batch Processing Optimization

from src.utils.database import DatabaseManager

db = DatabaseManager()

# Batch insert embeddings
embeddings_batch = [
    (binary_hash, code_snippet, embedding, metadata)
    for binary_hash, code_snippet, embedding, metadata in data
]

db.batch_insert_embeddings(embeddings_batch, batch_size=1000)

Cache Optimization

from src.utils.multi_level_cache import MultiLevelCache

cache = MultiLevelCache(
    l1_size=1000,           # In-memory cache size
    l2_ttl=3600,            # Redis TTL (1 hour)
    l3_ttl=86400            # PostgreSQL TTL (1 day)
)

# Warm up cache
cache.warm_up(frequently_accessed_keys)

Troubleshooting

Common Issues

Issue: "OPENROUTER_API_KEY not found"

Solution: Set environment variable or pass explicitly:

export OPENROUTER_API_KEY=sk-or-v1-your-key
# or
python -c "from src.agents.orchestrator import OrchestratingAgent; oa = OrchestratingAgent(openrouter_api_key='sk-or-v1-your-key')"

Issue: "PostgreSQL connection refused"

Solution: Verify PostgreSQL is running and accessible:

# Check PostgreSQL status
docker-compose ps postgres

# Verify connection
psql -h localhost -U raverse -d raverse_db -c "SELECT 1"

# Check logs
docker-compose logs postgres

Issue: "Redis connection timeout"

Solution: Verify Redis is running:

# Check Redis status
docker-compose ps redis

# Test connection
redis-cli -h localhost -p 6379 ping

# Check logs
docker-compose logs redis

Issue: "Vector search returns no results"

Solution: Verify embeddings are generated and indexed:

-- Check embedding count
SELECT COUNT(*) FROM code_embeddings WHERE embedding IS NOT NULL;

-- Check index status
SELECT * FROM pg_indexes WHERE tablename = 'code_embeddings';

-- Verify similarity threshold
SELECT 1 - (embedding <=> query_embedding::vector) AS similarity
FROM code_embeddings
LIMIT 1;

Issue: "Agent execution timeout"

Solution: Increase timeout or optimize agent:

# Increase timeout
oa = OrchestratingAgent(timeout=60)  # 60 seconds

# Or optimize agent logic
# - Reduce binary size
# - Use cached results
# - Increase LLM timeout

Issue: "Out of memory during analysis"

Solution: Reduce memory usage:

# Use lighter memory preset
from src.config.agent_memory_config import MEMORY_PRESETS
preset = MEMORY_PRESETS['light']

# Or reduce batch sizes
embedding_gen = EmbeddingGenerator(batch_size=8)  # Reduce from 32

# Or use streaming for large files

Debug Mode

Enable debug logging:

import logging

# Set debug level
logging.basicConfig(level=logging.DEBUG)

# Or for specific module
logger = logging.getLogger('src.agents.orchestrator')
logger.setLevel(logging.DEBUG)

Performance Profiling

Profile agent execution:

import cProfile
import pstats
from src.agents.orchestrator import OrchestratingAgent

# Profile binary analysis
profiler = cProfile.Profile()
profiler.enable()

oa = OrchestratingAgent()
result = oa.run("path/to/binary.exe")

profiler.disable()
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(20)  # Top 20 functions

Database Debugging

Query database for debugging:

-- Check binary analysis status
SELECT id, file_name, status, created_at FROM raverse.binaries
ORDER BY created_at DESC LIMIT 10;

-- Check analysis results
SELECT binary_id, result_data FROM raverse.analysis_results
WHERE binary_id = 123;

-- Check vector search index health
SELECT schemaname, tablename, indexname, idx_scan, idx_tup_read, idx_tup_fetch
FROM pg_stat_user_indexes
WHERE tablename LIKE '%embedding%';

-- Check cache efficiency
SELECT COUNT(*) as total_queries,
       SUM(CASE WHEN cached THEN 1 ELSE 0 END) as cached_queries,
       ROUND(100.0 * SUM(CASE WHEN cached THEN 1 ELSE 0 END) / COUNT(*), 2) as cache_hit_ratio
FROM query_log;

Development

Running Tests

# PowerShell
.\scripts\run_tests.ps1 -Verbose -Coverage

# Bash
bash scripts/run_tests.sh --verbose --coverage

Code Quality

# Format code
black src/ tests/

# Type checking
mypy src/

# Linting
ruff check src/

API Reference

Orchestrator API

OrchestratingAgent (Offline)

class OrchestratingAgent:
    def __init__(self, openrouter_api_key=None, model=None, use_database=True)
    def run(self, binary_path: str) -> Dict
    def call_openrouter(self, prompt: str, max_tokens: int = 500) -> Dict

Methods:

• run(binary_path): Execute complete offline pipeline

  • Returns: Analysis result with patches, verification status
  • Raises: ValueError if API key missing, FileNotFoundError if binary not found

• call_openrouter(prompt, max_tokens): Call OpenRouter LLM API

  • Returns: JSON response from LLM
  • Caches responses in Redis/PostgreSQL
  • Implements exponential backoff retry

OnlineOrchestrationAgent (Online)

class OnlineOrchestrationAgent:
    def __init__(self, api_key: str, model: str)
    def execute(self, target_url: str, scope: Dict, options: Dict) -> Dict
    def _execute_pipeline(self, target_url: str, scope: Dict, options: Dict) -> Dict

Methods:

• execute(target_url, scope, options): Execute complete online pipeline
  • Returns: Pipeline results with all agent outputs
  • Phases: Recon → Traffic → JS → API → WASM → Security → Validation → Reporting

Database API

DatabaseManager

class DatabaseManager:
    def __init__(self, host='localhost', port=5432, user='raverse', password='', database='raverse_db')
    def get_connection(self) -> Connection
    def create_binary_record(self, file_name, file_path, file_hash, file_size, file_type, architecture, metadata) -> int
    def search_similar_instructions(self, embedding: List[float], limit: int = 10) -> List[Dict]
    def execute_query(self, query: str, params: Tuple = ()) -> List[Dict]

Methods:

• create_binary_record(): Store binary metadata

  • Returns: Binary ID
  • Handles duplicates with ON CONFLICT

• search_similar_instructions(): Vector similarity search

  • Returns: List of similar instructions with similarity scores
  • Uses HNSW index for performance

• execute_query(): Execute arbitrary SQL

  • Returns: Query results as list of dicts
  • Implements connection pooling and retry logic

Cache API

CacheManager

class CacheManager:
    def __init__(self, redis_host='localhost', redis_port=6379)
    def cache_analysis(self, binary_hash: str, analysis_type: str, result: Dict) -> None
    def get_cached_analysis(self, binary_hash: str, analysis_type: str = 'full_analysis') -> Optional[Dict]
    def cache_llm_response(self, prompt: str, model: str, response: Dict) -> None
    def get_cached_llm_response(self, prompt: str, model: str) -> Optional[Dict]

Methods:

• cache_analysis(): Cache analysis results

  • TTL: 7 days for analysis results
  • Key format: analysis:{binary_hash}:{type}

• get_cached_analysis(): Retrieve cached analysis

  • Returns: Cached result or None if expired/missing

• cache_llm_response(): Cache LLM API responses

  • TTL: 24 hours for LLM responses
  • Key format: llm:{hash(prompt)}:{model}

Embedding API

EmbeddingGenerator

class EmbeddingGenerator:
    def __init__(self, model_name='all-MiniLM-L6-v2', batch_size=32, cache_manager=None)
    def generate_embedding(self, text: str) -> np.ndarray
    def generate_code_embedding(self, code: str) -> np.ndarray
    def batch_encode(self, texts: List[str], show_progress_bar=False) -> np.ndarray

Methods:

• generate_embedding(): Generate text embedding

  • Returns: 384-dimensional numpy array
  • Cached in Redis (TTL: 7 days)

• generate_code_embedding(): Generate code-specific embedding

  • Returns: 384-dimensional numpy array
  • Optimized for code similarity

• batch_encode(): Batch embedding generation

  • Returns: 2D numpy array (n_texts, 384)
  • Efficient batch processing with caching

Semantic Search API

SemanticSearchEngine

class SemanticSearchEngine:
    def __init__(self, db_manager: DatabaseManager, cache_manager: CacheManager)
    def store_code_embedding(self, binary_hash: str, code_snippet: str, metadata: Dict = None) -> int
    def find_similar_code(self, query: str, limit: int = 10, similarity_threshold: float = 0.7) -> List[Dict]
    def search_by_pattern(self, pattern: str, limit: int = 10) -> List[Dict]

Methods:

• store_code_embedding(): Store code with embedding

  • Returns: Embedding ID
  • Stores in PostgreSQL with pgvector

• find_similar_code(): Find similar code snippets

  • Returns: List of similar code with similarity scores
  • Filters by similarity_threshold

• search_by_pattern(): Pattern-based search

  • Returns: Matching code snippets
  • Uses regex or pattern matching

Integration Guide

Integrating with External Systems

Webhook Integration

from flask import Flask, request
from src.agents.orchestrator import OrchestratingAgent

app = Flask(__name__)
oa = OrchestratingAgent()

@app.route('/analyze', methods=['POST'])
def analyze_binary():
    """Webhook endpoint for binary analysis."""
    binary_path = request.json.get('binary_path')

    try:
        result = oa.run(binary_path)
        return {
            'status': 'success',
            'result': result
        }, 200
    except Exception as e:
        return {
            'status': 'error',
            'message': str(e)
        }, 500

Message Queue Integration

import pika
import json
from src.agents.orchestrator import OrchestratingAgent

# Connect to RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='binary_analysis')

oa = OrchestratingAgent()

def callback(ch, method, properties, body):
    """Process binary analysis from queue."""
    message = json.loads(body)
    binary_path = message['binary_path']

    result = oa.run(binary_path)

    # Publish result
    channel.basic_publish(
        exchange='',
        routing_key='analysis_results',
        body=json.dumps(result)
    )

    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='binary_analysis', on_message_callback=callback)
channel.start_consuming()

REST API Integration

from fastapi import FastAPI, File, UploadFile
from src.agents.orchestrator import OrchestratingAgent
import tempfile
import os

app = FastAPI()
oa = OrchestratingAgent()

@app.post("/api/v1/analyze")
async def analyze_binary(file: UploadFile = File(...)):
    """REST API endpoint for binary analysis."""

    # Save uploaded file
    with tempfile.NamedTemporaryFile(delete=False) as tmp:
        contents = await file.read()
        tmp.write(contents)
        tmp_path = tmp.name

    try:
        # Analyze binary
        result = oa.run(tmp_path)
        return {
            'status': 'success',
            'analysis': result
        }
    finally:
        # Clean up
        os.unlink(tmp_path)

@app.get("/api/v1/status/{analysis_id}")
async def get_analysis_status(analysis_id: str):
    """Get analysis status."""
    from src.utils.database import DatabaseManager

    db = DatabaseManager()
    result = db.execute_query(
        "SELECT status FROM raverse.binaries WHERE id = %s",
        (analysis_id,)
    )

    if result:
        return {'status': result[0]['status']}
    return {'error': 'Analysis not found'}, 404

Kubernetes Deployment Integration

apiVersion: apps/v1
kind: Deployment
metadata:
  name: raverse-analyzer
spec:
  replicas: 3
  selector:
    matchLabels:
      app: raverse-analyzer
  template:
    metadata:
      labels:
        app: raverse-analyzer
    spec:
      containers:
      - name: raverse
        image: raverse:latest
        env:
        - name: OPENROUTER_API_KEY
          valueFrom:
            secretKeyRef:
              name: raverse-secrets
              key: api-key
        - name: DB_HOST
          value: postgres-service
        - name: REDIS_HOST
          value: redis-service
        resources:
          requests:
            memory: "4Gi"
            cpu: "2"
          limits:
            memory: "8Gi"
            cpu: "4"
        ports:
        - containerPort: 8000

Agent Implementation Details

Offline Binary Analysis Agents

DisassemblyAnalysisAgent (DAA)

Purpose: Extract and analyze binary structure

Implementation (src/agents/disassembly_agent.py):

class DisassemblyAnalysisAgent:
    def __init__(self, openrouter_agent):
        self.openrouter_agent = openrouter_agent
        self.analyzer = BinaryAnalyzer()

    def disassemble(self, binary_path: str) -> Dict:
        """Disassemble binary and extract functions."""
        # Extract metadata
        metadata = self.analyzer.extract_metadata(binary_path)

        # Disassemble using Capstone
        disassembly = self.analyzer.disassemble(binary_path)

        # Identify functions
        functions = self.analyzer.identify_functions(disassembly)

        # Generate embeddings for semantic search
        embeddings = self._generate_embeddings(disassembly)

        return {
            'metadata': metadata,
            'disassembly': disassembly,
            'functions': functions,
            'embeddings': embeddings
        }

Input: Binary file path Output: Disassembly, functions, metadata, embeddings Model: Capstone disassembly engine Performance: 2-5 seconds for typical binaries

LogicIdentificationMappingAgent (LIMA)

Purpose: Analyze control flow and data flow

Implementation (src/agents/logic_identification.py):

class LogicIdentificationMappingAgent:
    def __init__(self, openrouter_agent):
        self.openrouter_agent = openrouter_agent

    def identify_logic(self, daa_output: Dict) -> Dict:
        """Identify logic and generate mapping."""
        # Analyze control flow
        control_flow = self._analyze_control_flow(daa_output)

        # Analyze data flow
        data_flow = self._analyze_data_flow(daa_output)

        # Identify algorithms
        algorithms = self._identify_algorithms(control_flow, data_flow)

        # Generate flowchart
        flowchart = self._generate_flowchart(control_flow)

        # Use LLM for semantic analysis
        llm_analysis = self.openrouter_agent.call_openrouter(
            f"Analyze this binary logic: {control_flow}"
        )

        return {
            'control_flow': control_flow,
            'data_flow': data_flow,
            'algorithms': algorithms,
            'flowchart': flowchart,
            'llm_analysis': llm_analysis
        }

Input: DAA output (disassembly, functions) Output: Logic maps, control/data flow, algorithms Model: OpenRouter LLM Performance: 3-8 seconds

PatchingExecutionAgent (PEA)

Purpose: Generate and apply binary patches

Implementation (src/agents/patching_execution.py):

class PatchingExecutionAgent:
    def __init__(self, openrouter_agent):
        self.openrouter_agent = openrouter_agent

    def patch_binary(self, lima_output: Dict, binary_path: str) -> str:
        """Apply patches to binary."""
        # Extract patch information
        jump_addr = lima_output.get('jump_addr')
        opcode = lima_output.get('opcode')

        # Create backup
        backup_path = f"{binary_path}.backup"
        shutil.copy2(binary_path, backup_path)

        # Convert virtual address to file offset
        file_offset = self._va_to_file_offset(binary_path, jump_addr)

        # Apply patch
        with open(binary_path, 'r+b') as f:
            f.seek(file_offset)
            f.write(bytes.fromhex(opcode))

        return binary_path

Input: LIMA output (logic maps), binary path Output: Patched binary file path Model: Binary utilities Performance: 1-2 seconds

VerificationAgent (VA)

Purpose: Validate patch integrity and functionality

Implementation (src/agents/verification.py):

class VerificationAgent:
    def __init__(self, openrouter_agent):
        self.openrouter_agent = openrouter_agent

    def verify_patch(self, pea_output: str, original_binary: str) -> Dict:
        """Verify patch integrity."""
        # Verify binary structure
        structure_valid = self._verify_structure(pea_output)

        # Verify patch was applied
        patch_applied = self._verify_patch_applied(pea_output, original_binary)

        # Test functionality
        functionality_ok = self._test_functionality(pea_output)

        # Generate verification report
        report = {
            'structure_valid': structure_valid,
            'patch_applied': patch_applied,
            'functionality_ok': functionality_ok,
            'success': all([structure_valid, patch_applied, functionality_ok])
        }

        return report

Input: Patched binary path, original binary path Output: Verification report with success status Model: Binary analysis Performance: 2-3 seconds

Online Analysis Agents

ReconnaissanceAgent

Purpose: Discover target technology stack and endpoints

Key Features:

• Technology stack detection (frameworks, libraries, versions)
• Endpoint discovery (URLs, API paths)
• Server information gathering
• DNS enumeration

Implementation Pattern:

class ReconnaissanceAgent(OnlineBaseAgent):
    def _execute_impl(self, task: Dict) -> Dict:
        target_url = task.get('target_url')

        # Detect technologies
        tech_stack = self._detect_technologies(target_url)

        # Discover endpoints
        endpoints = self._discover_endpoints(target_url)

        # Gather server info
        server_info = self._gather_server_info(target_url)

        return {
            'technologies': tech_stack,
            'endpoints': endpoints,
            'server_info': server_info
        }

TrafficInterceptionAgent

Purpose: Capture and analyze HTTP(S) traffic

Key Features:

• HTTPS traffic interception (with mitmproxy)
• Request/response analysis
• API call extraction
• Pattern detection

Implementation Pattern:

class TrafficInterceptionAgent(OnlineBaseAgent):
    def _execute_impl(self, task: Dict) -> Dict:
        target_url = task.get('target_url')
        duration = task.get('duration_seconds', 60)

        # Start traffic capture
        captured_traffic = self._capture_traffic(target_url, duration)

        # Analyze traffic
        api_calls = self._extract_api_calls(captured_traffic)
        patterns = self._detect_patterns(captured_traffic)

        return {
            'traffic': captured_traffic,
            'api_calls': api_calls,
            'patterns': patterns
        }

JavaScriptAnalysisAgent

Purpose: Deobfuscate and analyze JavaScript code

Key Features:

• JavaScript deobfuscation
• API call extraction from JS
• Client-side logic analysis
• Dependency detection

Implementation Pattern:

class JavaScriptAnalysisAgent(OnlineBaseAgent):
    def _execute_impl(self, task: Dict) -> Dict:
        js_code = task.get('javascript_code')

        # Deobfuscate JavaScript
        deobfuscated = self._deobfuscate(js_code)

        # Extract API calls
        api_calls = self._extract_api_calls(deobfuscated)

        # Analyze logic
        logic_analysis = self.orchestrator.call_openrouter(
            f"Analyze this JavaScript: {deobfuscated}"
        )

        return {
            'deobfuscated_code': deobfuscated,
            'api_calls': api_calls,
            'logic_analysis': logic_analysis
        }

APIReverseEngineeringAgent

Purpose: Map API endpoints and generate documentation

Key Features:

• Endpoint mapping
• OpenAPI spec generation
• Parameter extraction
• Authentication detection

Implementation Pattern:

class APIReverseEngineeringAgent(OnlineBaseAgent):
    def _execute_impl(self, task: Dict) -> Dict:
        traffic_data = task.get('traffic_data')

        # Extract endpoints
        endpoints = self._extract_endpoints(traffic_data)

        # Generate OpenAPI spec
        openapi_spec = self._generate_openapi_spec(endpoints)

        # Detect authentication
        auth_methods = self._detect_authentication(traffic_data)

        return {
            'endpoints': endpoints,
            'openapi_spec': openapi_spec,
            'authentication': auth_methods
        }

Advanced Agents

RAGOrchestratorAgent

Purpose: Retrieve-augmented generation for intelligent analysis

Implementation Pattern:

class RAGOrchestratorAgent(OnlineBaseAgent):
    def _execute_impl(self, task: Dict) -> Dict:
        query = task.get('query')
        context = task.get('context')

        # Generate query embedding
        query_embedding = self._generate_embedding(query)

        # Retrieve relevant knowledge
        retrieved_knowledge = self._retrieve_knowledge(
            query_embedding,
            limit=5,
            threshold=0.7
        )

        # Augment prompt with retrieved knowledge
        augmented_prompt = self._augment_prompt(query, retrieved_knowledge)

        # Generate response
        response = self.orchestrator.call_openrouter(augmented_prompt)

        return {
            'query': query,
            'retrieved_knowledge': retrieved_knowledge,
            'generated_response': response,
            'confidence': self._calculate_confidence(retrieved_knowledge)
        }

KnowledgeBaseAgent

Purpose: Manage knowledge base and embeddings

Implementation Pattern:

class KnowledgeBaseAgent(OnlineBaseAgent):
    def _execute_impl(self, task: Dict) -> Dict:
        action = task.get('action')  # 'store', 'retrieve', 'search'

        if action == 'store':
            # Store knowledge with embedding
            knowledge_id = self._store_knowledge(
                content=task.get('content'),
                metadata=task.get('metadata')
            )
            return {'knowledge_id': knowledge_id}

        elif action == 'retrieve':
            # Retrieve knowledge by ID
            knowledge = self._retrieve_knowledge_by_id(task.get('knowledge_id'))
            return {'knowledge': knowledge}

        elif action == 'search':
            # Search knowledge by similarity
            results = self._search_knowledge(
                query=task.get('query'),
                limit=task.get('limit', 10)
            )
            return {'results': results}

Performance Metrics

• Binary Analysis: ~2-5 seconds per binary (depending on size)
• API Discovery: ~10-30 seconds per target
• Memory Usage: ~500MB-2GB (depending on cache settings)
• Database Queries: <100ms average (with pgvector indexing)
• Vector Search: <100ms p95 latency with HNSW index
• Embedding Generation: ~100 embeddings/second
• LLM API Calls: 1-5 seconds (depends on model and complexity)

Production Deployment Guide

Pre-Deployment Checklist

• All environment variables configured in .env
• PostgreSQL database initialized with schema
• Redis instance running and accessible
• OpenRouter API key validated
• SSL certificates configured (for HTTPS)
• Backup strategy implemented
• Monitoring and alerting configured
• Log aggregation set up
• Rate limiting configured
• Security scanning completed

Docker Compose Production Setup

version: '3.8'
services:
  postgres:
    image: pgvector/pgvector:pg17-latest
    environment:
      POSTGRES_USER: raverse
      POSTGRES_PASSWORD: ${DB_PASSWORD}
      POSTGRES_DB: raverse_db
    volumes:
      - postgres_data:/var/lib/postgresql/data
      - ./docker/postgres/init:/docker-entrypoint-initdb.d
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U raverse"]
      interval: 10s
      timeout: 5s
      retries: 5
    restart: unless-stopped

  redis:
    image: redis:8.2-alpine
    command: redis-server --appendonly yes --requirepass ${REDIS_PASSWORD}
    volumes:
      - redis_data:/data
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s
      timeout: 5s
      retries: 5
    restart: unless-stopped

  raverse:
    build:
      context: .
      dockerfile: Dockerfile
    environment:
      OPENROUTER_API_KEY: ${OPENROUTER_API_KEY}
      DB_HOST: postgres
      DB_PORT: 5432
      DB_USER: raverse
      DB_PASSWORD: ${DB_PASSWORD}
      DB_NAME: raverse_db
      REDIS_HOST: redis
      REDIS_PORT: 6379
      REDIS_PASSWORD: ${REDIS_PASSWORD}
      LOG_LEVEL: INFO
    depends_on:
      postgres:
        condition: service_healthy
      redis:
        condition: service_healthy
    volumes:
      - ./logs:/app/logs
      - ./results:/app/results
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./docker/prometheus/prometheus.yml:/etc/prometheus/prometheus.yml
      - prometheus_data:/prometheus
    ports:
      - "9090:9090"
    restart: unless-stopped

  grafana:
    image: grafana/grafana:latest
    environment:
      GF_SECURITY_ADMIN_PASSWORD: ${GRAFANA_PASSWORD}
    volumes:
      - grafana_data:/var/lib/grafana
      - ./docker/grafana/provisioning:/etc/grafana/provisioning
    ports:
      - "3000:3000"
    restart: unless-stopped

volumes:
  postgres_data:
  redis_data:
  prometheus_data:
  grafana_data:

Database Backup Strategy

#!/bin/bash
# backup.sh - Daily PostgreSQL backup

BACKUP_DIR="/backups/raverse"
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
BACKUP_FILE="$BACKUP_DIR/raverse_db_$TIMESTAMP.sql.gz"

# Create backup
pg_dump -h $DB_HOST -U $DB_USER -d $DB_NAME | gzip > $BACKUP_FILE

# Keep only last 30 days
find $BACKUP_DIR -name "raverse_db_*.sql.gz" -mtime +30 -delete

# Upload to S3
aws s3 cp $BACKUP_FILE s3://raverse-backups/

Kubernetes Deployment

apiVersion: apps/v1
kind: Deployment
metadata:
  name: raverse
spec:
  replicas: 3
  selector:
    matchLabels:
      app: raverse
  template:
    metadata:
      labels:
        app: raverse
    spec:
      containers:
      - name: raverse
        image: raverse:latest
        env:
        - name: OPENROUTER_API_KEY
          valueFrom:
            secretKeyRef:
              name: raverse-secrets
              key: api-key
        - name: DB_HOST
          value: postgres-service
        - name: REDIS_HOST
          value: redis-service
        resources:
          requests:
            memory: "4Gi"
            cpu: "2"
          limits:
            memory: "8Gi"
            cpu: "4"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 5

Monitoring Setup

Prometheus Metrics:

• agent_execution_duration_seconds: Agent execution time
• agent_execution_errors_total: Failed executions
• database_query_duration_seconds: Query performance
• cache_hit_ratio: Cache efficiency
• vector_search_duration_seconds: Search latency

Grafana Dashboards:

1. System Overview (CPU, memory, disk)
2. Agent Performance (execution times, success rates)
3. Database Metrics (query performance, connections)
4. Cache Efficiency (hit ratio, evictions)
5. API Discovery (crawl progress, endpoints)

Security Hardening

• Enable SSL/TLS for all connections
• Use strong database passwords (min 32 characters)
• Implement network policies for pod-to-pod communication
• Store secrets in Kubernetes Secrets or HashiCorp Vault
• Enable audit logging for all API calls
• Implement rate limiting on API endpoints
• Use read-only file systems where possible
• Scan container images for vulnerabilities

Scaling Configuration

Horizontal Scaling:

• Deploy multiple agent worker pods
• Use load balancer for traffic distribution
• Scale PostgreSQL with read replicas
• Scale Redis with cluster mode

Vertical Scaling:

• Increase CPU cores for LLM inference
• Increase RAM for caching (up to 16GB recommended)
• Use SSD storage for database
• Increase network bandwidth

Disaster Recovery

Component	RTO	RPO	Strategy
Application	5 min	0 min	Kubernetes auto-restart
Database	15 min	1 hour	Backup + replica promotion
Cache	5 min	0 min	Rebuild from database
Config	5 min	0 min	Version control + secrets

Security Considerations

⚠️ IMPORTANT: Use only on binaries and systems you own or are authorized to analyze.

• All API keys must be stored in .env (never commit to git)
• Database credentials should use strong passwords in production
• Enable SSL/TLS for remote deployments
• Use network isolation for sensitive analysis

Contributing

Contributions are welcome! Please:

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit changes (git commit -m 'Add amazing feature')
4. Push to branch (git push origin feature/amazing-feature)
5. Open a Pull Request

License

This project is licensed under the MIT License - see the LICENSE file for details.

Support

For issues, questions, or suggestions:

1. Check existing documentation
2. Review archived reports for historical context
3. Open an issue on GitHub
4. Contact the development team

Best Practices & Optimization

Code Organization Best Practices

Agent Development

# ✓ GOOD: Clear separation of concerns
class MyAgent(OnlineBaseAgent):
    def __init__(self, orchestrator, api_key, model):
        super().__init__(name="MyAgent", orchestrator=orchestrator,
                        api_key=api_key, model=model)
        self.db = DatabaseManager()
        self.cache = CacheManager()

    def _execute_impl(self, task: Dict) -> Dict:
        """Implement agent logic."""
        # Validate input
        if not self._validate_input(task):
            return {'error': 'Invalid input'}

        # Check cache
        cached = self.cache.get(task['id'])
        if cached:
            return cached

        # Execute logic
        result = self._process(task)

        # Cache result
        self.cache.set(task['id'], result, ttl=3600)

        return result

    def _validate_input(self, task: Dict) -> bool:
        """Validate input parameters."""
        required_fields = ['id', 'data']
        return all(field in task for field in required_fields)

    def _process(self, task: Dict) -> Dict:
        """Process task logic."""
        # Implementation here
        pass

Error Handling

# ✓ GOOD: Comprehensive error handling
try:
    result = oa.run(binary_path)
except FileNotFoundError:
    logger.error(f"Binary not found: {binary_path}")
    return {'error': 'Binary not found'}
except ValueError as e:
    logger.error(f"Invalid input: {e}")
    return {'error': str(e)}
except Exception as e:
    logger.exception(f"Unexpected error: {e}")
    return {'error': 'Internal server error'}

Database Optimization

Connection Pooling

# ✓ GOOD: Proper connection pooling
from sqlalchemy import create_engine

engine = create_engine(
    f"postgresql://{user}:{password}@{host}:{port}/{database}",
    pool_size=10,
    max_overflow=20,
    pool_recycle=3600,
    pool_pre_ping=True
)

Query Optimization

# ✓ GOOD: Efficient queries with indexes
# Create indexes for frequently searched columns
CREATE INDEX idx_binary_hash ON raverse.binaries(file_hash);
CREATE INDEX idx_embedding_hnsw ON raverse.code_embeddings
  USING hnsw (embedding vector_cosine_ops);

# Use EXPLAIN to analyze queries
EXPLAIN ANALYZE
SELECT * FROM code_embeddings
WHERE 1 - (embedding <=> query_embedding::vector) >= 0.7
ORDER BY embedding <=> query_embedding::vector
LIMIT 10;

Batch Operations

# ✓ GOOD: Batch inserts for performance
def batch_insert_embeddings(embeddings_list, batch_size=1000):
    """Insert embeddings in batches."""
    for i in range(0, len(embeddings_list), batch_size):
        batch = embeddings_list[i:i+batch_size]
        db.execute_many(
            "INSERT INTO code_embeddings (binary_hash, code, embedding) VALUES (%s, %s, %s)",
            batch
        )

Caching Strategy

Multi-Level Caching

# ✓ GOOD: Multi-level cache hierarchy
class MultiLevelCache:
    def __init__(self):
        self.l1_cache = {}  # In-memory (fast, limited)
        self.l2_cache = redis_client  # Redis (medium, distributed)
        self.l3_cache = db  # PostgreSQL (slow, persistent)

    def get(self, key):
        # Try L1 first
        if key in self.l1_cache:
            return self.l1_cache[key]

        # Try L2
        value = self.l2_cache.get(key)
        if value:
            self.l1_cache[key] = value
            return value

        # Try L3
        value = self.l3_cache.get(key)
        if value:
            self.l2_cache.set(key, value, ttl=3600)
            self.l1_cache[key] = value
            return value

        return None

Cache Invalidation

# ✓ GOOD: Proper cache invalidation
def update_binary_analysis(binary_id, new_result):
    """Update analysis and invalidate cache."""
    # Update database
    db.update_analysis(binary_id, new_result)

    # Invalidate caches
    cache_key = f"analysis:{binary_id}"
    redis_client.delete(cache_key)

    # Notify other services
    publish_event('analysis_updated', {'binary_id': binary_id})

Performance Tuning

Async Operations

# ✓ GOOD: Async operations for I/O
import asyncio

async def analyze_multiple_binaries(binary_paths):
    """Analyze multiple binaries concurrently."""
    tasks = [
        asyncio.create_task(analyze_binary_async(path))
        for path in binary_paths
    ]
    results = await asyncio.gather(*tasks)
    return results

async def analyze_binary_async(binary_path):
    """Async binary analysis."""
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(None, oa.run, binary_path)

Batch Processing

# ✓ GOOD: Batch processing for efficiency
def process_embeddings_batch(texts, batch_size=32):
    """Process embeddings in batches."""
    embeddings = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        batch_embeddings = embedding_gen.batch_encode(batch)
        embeddings.extend(batch_embeddings)
    return embeddings

Monitoring Best Practices

Logging

# ✓ GOOD: Structured logging
import logging
import json

logger = logging.getLogger(__name__)

def log_analysis(binary_id, status, duration_ms):
    """Log analysis with structured format."""
    logger.info(json.dumps({
        'event': 'analysis_complete',
        'binary_id': binary_id,
        'status': status,
        'duration_ms': duration_ms,
        'timestamp': datetime.utcnow().isoformat()
    }))

Metrics Collection

# ✓ GOOD: Prometheus metrics
from prometheus_client import Counter, Histogram, Gauge

analysis_duration = Histogram(
    'analysis_duration_seconds',
    'Time to complete analysis',
    buckets=(1, 2, 5, 10, 30, 60)
)

analysis_errors = Counter(
    'analysis_errors_total',
    'Total analysis errors'
)

cache_hit_ratio = Gauge(
    'cache_hit_ratio',
    'Cache hit ratio'
)

# Use in code
with analysis_duration.time():
    result = oa.run(binary_path)

Security Best Practices

Input Validation

# ✓ GOOD: Comprehensive input validation
def validate_binary_path(path):
    """Validate binary path."""
    # Check path exists
    if not os.path.exists(path):
        raise FileNotFoundError(f"Binary not found: {path}")

    # Check path is file
    if not os.path.isfile(path):
        raise ValueError(f"Path is not a file: {path}")

    # Check path is within allowed directory
    allowed_dir = os.path.abspath('/binaries')
    real_path = os.path.abspath(path)
    if not real_path.startswith(allowed_dir):
        raise ValueError(f"Path outside allowed directory: {path}")

    return real_path

Secrets Management

# ✓ GOOD: Secure secrets handling
import os
from dotenv import load_dotenv

# Load from .env file
load_dotenv()

# Get secrets from environment
api_key = os.getenv('OPENROUTER_API_KEY')
if not api_key:
    raise ValueError("OPENROUTER_API_KEY not set")

# Never log secrets
logger.info(f"Using API key: {api_key[:10]}...")  # Only show prefix

Testing Best Practices

Unit Tests

# ✓ GOOD: Comprehensive unit tests
import pytest

class TestOrchestrator:
    @pytest.fixture
    def orchestrator(self):
        return OrchestratingAgent(use_database=False)

    def test_run_success(self, orchestrator, tmp_path):
        """Test successful binary analysis."""
        # Create test binary
        binary_path = tmp_path / "test.bin"
        binary_path.write_bytes(b"test")

        # Run analysis
        result = orchestrator.run(str(binary_path))

        # Assert success
        assert result['success']
        assert 'patches' in result

    def test_run_invalid_path(self, orchestrator):
        """Test with invalid binary path."""
        with pytest.raises(FileNotFoundError):
            orchestrator.run("/nonexistent/binary")

Integration Tests

# ✓ GOOD: Integration tests with fixtures
@pytest.fixture
def db_session():
    """Create test database session."""
    db = DatabaseManager(database='raverse_test')
    db.create_tables()
    yield db
    db.drop_tables()

def test_end_to_end_analysis(db_session):
    """Test complete analysis pipeline."""
    # Setup
    binary_path = "tests/fixtures/test_binary.exe"

    # Execute
    oa = OrchestratingAgent(use_database=True)
    result = oa.run(binary_path)

    # Verify
    assert result['success']

    # Check database
    records = db_session.execute_query(
        "SELECT * FROM raverse.binaries WHERE file_hash = %s",
        (result['binary_hash'],)
    )
    assert len(records) > 0

Acknowledgments

• Built with Python 3.13+
• Powered by OpenRouter API
• Uses Capstone for binary disassembly
• Leverages PostgreSQL pgvector for semantic search
• Monitoring with Prometheus and Grafana
• Community contributions and feedback

---

Advanced Topics

Vector Database Optimization

HNSW Index Configuration

The Hierarchical Navigable Small World (HNSW) index is used for efficient vector similarity search:

-- Create HNSW index with optimal parameters
CREATE INDEX idx_embeddings_hnsw ON code_embeddings
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- Parameters explanation:
-- m = 16: Number of connections per node (higher = better quality, slower)
-- ef_construction = 64: Size of dynamic candidate list (higher = better quality, slower)

-- For production with large datasets:
CREATE INDEX idx_embeddings_hnsw_prod ON code_embeddings
USING hnsw (embedding vector_cosine_ops)
WITH (m = 32, ef_construction = 128);

-- Query-time parameter
SET hnsw.ef_search = 200;  -- Higher = more accurate, slower

Vector Similarity Metrics

RAVERSE uses cosine distance for similarity:

-- Cosine distance formula: 1 - (dot_product / (norm_a * norm_b))
-- In pgvector: 1 - (embedding <=> query_embedding::vector)

-- Example: Find top-10 similar code snippets
SELECT
    id, code_snippet, metadata,
    1 - (embedding <=> query_embedding::vector) AS similarity_score
FROM code_embeddings
WHERE 1 - (embedding <=> query_embedding::vector) >= 0.7
ORDER BY embedding <=> query_embedding::vector
LIMIT 10;

-- Similarity score interpretation:
-- 1.0 = Identical
-- 0.8-1.0 = Very similar
-- 0.6-0.8 = Similar
-- 0.4-0.6 = Somewhat similar
-- <0.4 = Dissimilar

Embedding Dimensions

Different embedding models provide different dimensions:

Model	Dimensions	Use Case	Speed
all-MiniLM-L6-v2	384	General purpose, balanced	Fast
all-mpnet-base-v2	768	High quality, slower	Medium
all-MiniLM-L12-v2	384	Lightweight, fast	Very Fast
sentence-transformers/all-roberta-large-v1	1024	High quality	Slow

Binary Analysis Deep Dive

Disassembly Process

# Step 1: Load binary and extract metadata
binary = BinaryAnalyzer.load_binary(binary_path)
metadata = {
    'architecture': binary.arch,  # x86, x64, ARM, etc.
    'file_type': binary.file_type,  # ELF, PE, Mach-O
    'entry_point': binary.entry_point,
    'sections': binary.sections
}

# Step 2: Disassemble using Capstone
from capstone import *
md = Cs(CS_ARCH_X86, CS_MODE_64)
instructions = list(md.disasm(binary_code, binary.entry_point))

# Step 3: Identify functions
functions = BinaryAnalyzer.identify_functions(instructions)

# Step 4: Extract control flow
cfg = BinaryAnalyzer.build_control_flow_graph(functions)

# Step 5: Generate embeddings for semantic search
embeddings = embedding_gen.batch_encode([
    instr.mnemonic + ' ' + instr.op_str
    for instr in instructions
])

Patch Generation

# Patch format: (address, original_bytes, new_bytes)
patches = [
    (0x401000, b'\x90\x90\x90\x90', b'\xc3\x00\x00\x00'),  # NOP to RET
    (0x401010, b'\x74\x05', b'\xeb\x05'),  # JE to JMP
]

# Apply patches
for address, original, new in patches:
    file_offset = va_to_file_offset(binary_path, address)
    with open(binary_path, 'r+b') as f:
        f.seek(file_offset)
        current = f.read(len(original))
        if current == original:
            f.seek(file_offset)
            f.write(new)
        else:
            raise ValueError(f"Patch mismatch at {hex(address)}")

Online Analysis Deep Dive

Traffic Interception with mitmproxy

from mitmproxy import http
from mitmproxy.tools.main import mitmdump

class APIInterceptor:
    def __init__(self):
        self.captured_requests = []
        self.captured_responses = []

    def request(self, flow: http.HTTPFlow) -> None:
        """Intercept HTTP request."""
        self.captured_requests.append({
            'method': flow.request.method,
            'url': flow.request.url,
            'headers': dict(flow.request.headers),
            'body': flow.request.content,
            'timestamp': time.time()
        })

    def response(self, flow: http.HTTPFlow) -> None:
        """Intercept HTTP response."""
        self.captured_responses.append({
            'status_code': flow.response.status_code,
            'headers': dict(flow.response.headers),
            'body': flow.response.content,
            'timestamp': time.time()
        })

# Start interception
interceptor = APIInterceptor()
mitmdump(['-s', 'interceptor.py', '--mode', 'transparent'])

JavaScript Deobfuscation

# Deobfuscate JavaScript using js-beautify
import jsbeautifier

obfuscated_js = """
var _0x4e2a=['log','Hello'];
(function(_0x2d3a1c){
    var _0x4e2a1f=function(_0x2d3a1c){
        while(--_0x2d3a1c){
            _0x2d3a1c['push'](_0x2d3a1c['shift']());
        }
    };
    _0x4e2a1f(++_0x2d3a1c);
}(_0x4e2a,0x1a7));

var _0x4e2a=function(_0x2d3a1c,_0x4e2a1f){
    _0x2d3a1c=_0x2d3a1c-0x0;
    var _0x4e2a1f=_0x4e2a[_0x2d3a1c];
    return _0x4e2a1f;
};

console[_0x4e2a('0x0')](_0x4e2a('0x1'));
"""

# Beautify
beautified = jsbeautifier.beautify(obfuscated_js)
print(beautified)

RAG Implementation Details

Retrieval Process

# Step 1: Generate query embedding
query = "How to bypass authentication?"
query_embedding = embedding_gen.generate_embedding(query)

# Step 2: Search knowledge base
results = db.search_similar_instructions(
    embedding=query_embedding,
    limit=5
)

# Step 3: Rank results by relevance
ranked_results = sorted(
    results,
    key=lambda x: x['similarity'],
    reverse=True
)

# Step 4: Filter by threshold
filtered_results = [
    r for r in ranked_results
    if r['similarity'] >= 0.7
]

# Step 5: Augment prompt
context = "\n".join([
    f"- {r['content']} (similarity: {r['similarity']:.2%})"
    for r in filtered_results
])

augmented_prompt = f"""
Based on the following knowledge:
{context}

Answer this question: {query}
"""

# Step 6: Generate response
response = llm.generate(augmented_prompt)

Knowledge Base Management

# Store knowledge with metadata
knowledge_entry = {
    'content': 'Binary patching technique using NOP instructions',
    'metadata': {
        'category': 'patching',
        'difficulty': 'beginner',
        'tags': ['binary', 'patch', 'nop'],
        'source': 'documentation',
        'created_at': datetime.utcnow()
    },
    'embedding': embedding_gen.generate_embedding(content)
}

# Insert into database
db.execute_query("""
    INSERT INTO knowledge_base (content, metadata, embedding)
    VALUES (%s, %s, %s)
""", (
    knowledge_entry['content'],
    json.dumps(knowledge_entry['metadata']),
    knowledge_entry['embedding']
))

Memory Management

Hierarchical Memory Implementation

class HierarchicalMemory:
    def __init__(self, window_size=3):
        self.recent = []  # Recent messages
        self.important = []  # Important messages
        self.archived = []  # Archived messages
        self.window_size = window_size

    def add_message(self, message, importance=0.5):
        """Add message to memory."""
        self.recent.append({
            'content': message,
            'importance': importance,
            'timestamp': time.time()
        })

        # Promote important messages
        if importance > 0.8:
            self.important.append(self.recent.pop())

        # Archive old messages
        if len(self.recent) > self.window_size:
            self.archived.append(self.recent.pop(0))

    def get_context(self):
        """Get context for LLM."""
        context = []
        context.extend(self.recent)
        context.extend(self.important[:5])
        return context

Memory-Augmented Generation

class MemoryAugmentedAgent:
    def __init__(self):
        self.memory = HierarchicalMemory()
        self.llm = OpenRouterLLM()

    def execute(self, task):
        """Execute task with memory augmentation."""
        # Get memory context
        memory_context = self.memory.get_context()

        # Augment prompt with memory
        augmented_prompt = self._augment_prompt(task, memory_context)

        # Generate response
        response = self.llm.generate(augmented_prompt)

        # Store in memory
        self.memory.add_message(
            f"Task: {task}\nResponse: {response}",
            importance=0.7
        )

        return response

Performance Profiling

CPU Profiling

import cProfile
import pstats

# Profile binary analysis
profiler = cProfile.Profile()
profiler.enable()

oa = OrchestratingAgent()
result = oa.run("test_binary.exe")

profiler.disable()

# Print statistics
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(20)  # Top 20 functions

Memory Profiling

from memory_profiler import profile

@profile
def analyze_large_binary(binary_path):
    """Profile memory usage."""
    oa = OrchestratingAgent()
    result = oa.run(binary_path)
    return result

# Run with memory profiler
# python -m memory_profiler script.py

Database Query Profiling

-- Enable query logging
SET log_statement = 'all';
SET log_duration = on;
SET log_min_duration_statement = 100;  -- Log queries > 100ms

-- Analyze query plan
EXPLAIN ANALYZE
SELECT * FROM code_embeddings
WHERE 1 - (embedding <=> query_embedding::vector) >= 0.7
ORDER BY embedding <=> query_embedding::vector
LIMIT 10;

-- Check index usage
SELECT schemaname, tablename, indexname, idx_scan, idx_tup_read, idx_tup_fetch
FROM pg_stat_user_indexes
ORDER BY idx_scan DESC;

Case Studies

Case Study 1: Binary Vulnerability Analysis

Scenario: Analyze a vulnerable binary and generate patches

Process:

1. DAA disassembles binary and identifies vulnerable function
2. LIMA analyzes control flow and identifies vulnerability pattern
3. PEA generates patch (e.g., bounds check before buffer access)
4. VA verifies patch integrity and functionality

Results:

• Vulnerability identified in 3 seconds
• Patch generated in 2 seconds
• Verification completed in 1 second
• Total time: 6 seconds

Case Study 2: API Discovery

Scenario: Discover and document APIs in web application

Process:

1. Reconnaissance identifies technology stack
2. Traffic interception captures API calls
3. JavaScript analysis extracts client-side API calls
4. API reverse engineering generates OpenAPI spec

Results:

• 47 API endpoints discovered
• 89% accuracy in parameter extraction
• OpenAPI spec generated automatically
• Total time: 2 minutes

Case Study 3: RAG-Enhanced Analysis

Scenario: Analyze code with knowledge base augmentation

Process:

1. Query knowledge base for similar patterns
2. Retrieve top-5 relevant knowledge entries
3. Augment LLM prompt with retrieved knowledge
4. Generate analysis with context

Results:

• 23% improvement in analysis accuracy
• 15% reduction in hallucinations
• Better explanations with sources
• Confidence score: 0.92

Complete API Specifications

Orchestrator API Reference

OrchestratingAgent.run()

def run(self, binary_path: str) -> Dict[str, Any]:
    """
    Execute complete offline binary analysis pipeline.

    Args:
        binary_path (str): Path to binary file to analyze

    Returns:
        Dict with keys:
            - success (bool): Whether analysis succeeded
            - binary_id (int): Database ID of binary record
            - binary_hash (str): SHA256 hash of binary
            - metadata (Dict): Binary metadata (arch, type, size, etc.)
            - disassembly (Dict): Disassembly output from DAA
            - logic_map (Dict): Logic mapping from LIMA
            - patches (List): Generated patches from PEA
            - verification (Dict): Verification results from VA
            - execution_time_ms (int): Total execution time

    Raises:
        FileNotFoundError: If binary_path doesn't exist
        ValueError: If API key not configured
        RuntimeError: If analysis fails

    Example:
        >>> oa = OrchestratingAgent()
        >>> result = oa.run('/path/to/binary.exe')
        >>> print(f"Success: {result['success']}")
        >>> print(f"Patches: {len(result['patches'])}")
    """

OrchestratingAgent.call_openrouter()

def call_openrouter(self, prompt: str, max_tokens: int = 500) -> Dict[str, Any]:
    """
    Call OpenRouter LLM API with caching.

    Args:
        prompt (str): Prompt to send to LLM
        max_tokens (int): Maximum tokens in response (default: 500)

    Returns:
        Dict with keys:
            - content (str): LLM response text
            - model (str): Model used
            - tokens_used (int): Tokens consumed
            - cached (bool): Whether response was cached

    Raises:
        ValueError: If API key not set
        RuntimeError: If API call fails

    Example:
        >>> response = oa.call_openrouter("Analyze this code: ...")
        >>> print(response['content'])
    """

Database API Reference

DatabaseManager.search_similar_instructions()

def search_similar_instructions(
    self,
    embedding: List[float],
    limit: int = 10,
    threshold: float = 0.7
) -> List[Dict[str, Any]]:
    """
    Search for similar instructions using vector similarity.

    Args:
        embedding (List[float]): Query embedding (384 dimensions)
        limit (int): Maximum results to return (default: 10)
        threshold (float): Minimum similarity score (default: 0.7)

    Returns:
        List of dicts with keys:
            - address (str): Instruction address
            - instruction (str): Instruction text
            - opcode (str): Opcode bytes
            - operands (str): Operand text
            - similarity (float): Similarity score (0-1)

    Example:
        >>> embedding = embedding_gen.generate_embedding("cmp eax, 0")
        >>> results = db.search_similar_instructions(embedding, limit=5)
        >>> for r in results:
        ...     print(f"{r['instruction']} ({r['similarity']:.2%})")
    """

DatabaseManager.create_binary_record()

def create_binary_record(
    self,
    file_name: str,
    file_path: str,
    file_hash: str,
    file_size: int,
    file_type: str,
    architecture: str,
    metadata: Dict[str, Any] = None
) -> int:
    """
    Create binary record in database.

    Args:
        file_name (str): Binary filename
        file_path (str): Full path to binary
        file_hash (str): SHA256 hash
        file_size (int): File size in bytes
        file_type (str): File type (ELF, PE, Mach-O)
        architecture (str): Architecture (x86, x64, ARM)
        metadata (Dict): Additional metadata

    Returns:
        int: Binary ID in database

    Raises:
        IntegrityError: If binary already exists

    Example:
        >>> binary_id = db.create_binary_record(
        ...     file_name="app.exe",
        ...     file_path="/binaries/app.exe",
        ...     file_hash="abc123...",
        ...     file_size=1024000,
        ...     file_type="PE",
        ...     architecture="x64"
        ... )
    """

Cache API Reference

CacheManager.cache_analysis()

def cache_analysis(
    self,
    binary_hash: str,
    analysis_type: str,
    result: Dict[str, Any],
    ttl: int = 604800
) -> None:
    """
    Cache analysis result.

    Args:
        binary_hash (str): Binary SHA256 hash
        analysis_type (str): Type of analysis (full_analysis, quick_scan)
        result (Dict): Analysis result to cache
        ttl (int): Time to live in seconds (default: 7 days)

    Example:
        >>> cache.cache_analysis(
        ...     binary_hash="abc123...",
        ...     analysis_type="full_analysis",
        ...     result=analysis_result,
        ...     ttl=86400  # 1 day
        ... )
    """

CacheManager.get_cached_analysis()

def get_cached_analysis(
    self,
    binary_hash: str,
    analysis_type: str = 'full_analysis'
) -> Optional[Dict[str, Any]]:
    """
    Retrieve cached analysis result.

    Args:
        binary_hash (str): Binary SHA256 hash
        analysis_type (str): Type of analysis

    Returns:
        Dict with cached result, or None if not found/expired

    Example:
        >>> cached = cache.get_cached_analysis("abc123...")
        >>> if cached:
        ...     print("Using cached result")
        ... else:
        ...     print("Cache miss, running analysis")
    """

Configuration Reference

Environment Variables Complete List

API Configuration

# OpenRouter API
OPENROUTER_API_KEY=sk-or-v1-...          # Required: OpenRouter API key
OPENROUTER_MODEL=meta-llama/llama-3.3-70b-instruct:free  # LLM model to use
OPENROUTER_TIMEOUT=30                    # API timeout in seconds
OPENROUTER_MAX_RETRIES=3                 # Max retry attempts

Database Configuration

# PostgreSQL
DB_HOST=localhost                        # Database host
DB_PORT=5432                             # Database port
DB_USER=raverse                          # Database user
DB_PASSWORD=your_password                # Database password
DB_NAME=raverse_db                       # Database name
DB_POOL_SIZE=10                          # Connection pool size
DB_MAX_OVERFLOW=20                       # Max overflow connections
DB_POOL_RECYCLE=3600                     # Recycle connections after (seconds)

Cache Configuration

# Redis
REDIS_HOST=localhost                     # Redis host
REDIS_PORT=6379                          # Redis port
REDIS_DB=0                               # Redis database number
REDIS_PASSWORD=                          # Redis password (if required)
REDIS_CLUSTER_MODE=false                 # Enable cluster mode
REDIS_CACHE_TTL=604800                   # Cache TTL in seconds (7 days)

Logging Configuration

# Logging
LOG_LEVEL=INFO                           # Log level (DEBUG, INFO, WARNING, ERROR)
LOG_FILE=logs/raverse.log                # Log file path
LOG_FORMAT=json                          # Log format (json, text)
LOG_MAX_SIZE=104857600                   # Max log file size (100MB)
LOG_BACKUP_COUNT=10                      # Number of backup log files

Feature Configuration

# Features
ENABLE_VECTOR_SEARCH=true                # Enable vector similarity search
ENABLE_RAG=true                          # Enable RAG augmentation
ENABLE_CACHING=true                      # Enable result caching
ENABLE_MONITORING=true                   # Enable Prometheus metrics
ENABLE_PROFILING=false                   # Enable performance profiling

Performance Configuration

# Performance
BATCH_SIZE=32                            # Embedding batch size
MAX_CONCURRENT_ANALYSES=5                # Max concurrent analyses
EMBEDDING_CACHE_SIZE=10000               # In-memory embedding cache size
VECTOR_SEARCH_LIMIT=10                   # Default vector search limit
VECTOR_SEARCH_THRESHOLD=0.7              # Vector similarity threshold

Configuration Files Reference

src/config/settings.py

# Main application settings
DEBUG = False
ENVIRONMENT = 'production'
LOG_LEVEL = 'INFO'

# Database settings
DATABASE = {
    'host': 'localhost',
    'port': 5432,
    'user': 'raverse',
    'password': '',
    'database': 'raverse_db'
}

# Cache settings
CACHE = {
    'backend': 'redis',
    'host': 'localhost',
    'port': 6379,
    'ttl': 604800  # 7 days
}

# LLM settings
LLM = {
    'provider': 'openrouter',
    'model': 'meta-llama/llama-3.3-70b-instruct:free',
    'timeout': 30,
    'max_retries': 3
}

src/config/agent_memory_config.py

# Agent memory configurations
AGENT_MEMORY_CONFIG = {
    'version_manager': {
        'strategy': 'hierarchical',
        'preset': 'medium',
        'reason': 'Critical version info retention'
    },
    'knowledge_base': {
        'strategy': 'retrieval',
        'preset': 'heavy',
        'reason': 'Semantic search for knowledge'
    },
    'quality_gate': {
        'strategy': 'memory_augmented',
        'preset': 'medium',
        'reason': 'Critical metrics + context'
    },
    # ... more agents
}

# Memory presets
MEMORY_PRESETS = {
    'none': {
        'description': 'No memory',
        'ram_mb': 0,
        'cpu_percent': 0
    },
    'light': {
        'description': 'Sliding window memory',
        'ram_mb': 5,
        'cpu_percent': 1
    },
    'medium': {
        'description': 'Hierarchical memory',
        'ram_mb': 20,
        'cpu_percent': 3
    },
    'heavy': {
        'description': 'Retrieval + RAG',
        'ram_mb': 100,
        'cpu_percent': 5
    }
}

src/config/deepcrawler_config.py

# DeepCrawler configuration
DEEPCRAWLER_CONFIG = {
    'max_depth': 3,
    'max_urls': 10000,
    'max_concurrent': 5,
    'timeout': 30,
    'rate_limit': 20.0,  # requests per minute
    'detect_rest_apis': True,
    'detect_graphql': True,
    'detect_websockets': True,
    'min_confidence_score': 0.6,
    'output_format': 'openapi',
    'output_dir': './crawl_results'
}

Comprehensive Troubleshooting Guide

Database Issues

Issue: "FATAL: remaining connection slots reserved for non-replication superuser connections"

Cause: Connection pool exhausted

Solution:

# Increase pool size
db = DatabaseManager(
    pool_size=20,
    max_overflow=30
)

# Or check active connections
SELECT count(*) FROM pg_stat_activity;

# Kill idle connections
SELECT pg_terminate_backend(pid)
FROM pg_stat_activity
WHERE state = 'idle' AND query_start < now() - interval '1 hour';

Issue: "HNSW index search returns no results"

Cause: Index not built or threshold too high

Solution:

-- Check index exists
SELECT * FROM pg_indexes WHERE tablename = 'code_embeddings';

-- Rebuild index if needed
REINDEX INDEX idx_embeddings_hnsw;

-- Lower threshold
SELECT * FROM code_embeddings
WHERE 1 - (embedding <=> query_embedding::vector) >= 0.5
LIMIT 10;

Cache Issues

Issue: "Redis connection refused"

Cause: Redis not running or wrong host/port

Solution:

# Check Redis status
redis-cli ping

# Start Redis
docker-compose up -d redis

# Test connection
redis-cli -h localhost -p 6379 ping

Issue: "Cache hit ratio very low"

Cause: Cache TTL too short or cache size too small

Solution:

# Increase TTL
cache.cache_analysis(
    binary_hash="abc123",
    analysis_type="full_analysis",
    result=result,
    ttl=2592000  # 30 days instead of 7
)

# Increase cache size
cache = CacheManager(
    redis_host='localhost',
    redis_port=6379,
    max_memory='2gb'  # Increase from 1gb
)

Performance Issues

Issue: "Analysis taking too long"

Cause: Large binary or slow LLM model

Solution:

# Use faster model
oa = OrchestratingAgent(
    model="meta-llama/llama-3.2-3b-instruct:free"  # Faster
)

# Or reduce binary size
# Split large binary into chunks
chunks = split_binary(binary_path, chunk_size=1000000)
for chunk in chunks:
    result = oa.run(chunk)

Issue: "High memory usage"

Cause: Large embeddings or memory preset too high

Solution:

# Use lighter memory preset
from src.config.agent_memory_config import MEMORY_PRESETS
preset = MEMORY_PRESETS['light']

# Or reduce batch size
embedding_gen = EmbeddingGenerator(batch_size=8)

# Or use streaming
for chunk in stream_embeddings(texts, batch_size=16):
    process_chunk(chunk)

API Issues

Issue: "OpenRouter API rate limit exceeded"

Cause: Too many concurrent requests

Solution:

# Implement rate limiting
from ratelimit import limits, sleep_and_retry

@sleep_and_retry
@limits(calls=10, period=60)  # 10 calls per minute
def call_openrouter(prompt):
    return oa.call_openrouter(prompt)

# Or use queue
from queue import Queue
request_queue = Queue(maxsize=10)

Issue: "Invalid API key"

Cause: API key not set or invalid

Solution:

# Check environment variable
echo $OPENROUTER_API_KEY

# Set if missing
export OPENROUTER_API_KEY=sk-or-v1-your-key

# Verify key format
# Should start with: sk-or-v1-

Data Structures & Algorithms

Binary Analysis Data Structures

Instruction Representation

class Instruction:
    """Represents a single CPU instruction."""

    def __init__(self, address, opcode, mnemonic, operands):
        self.address = address  # Virtual address
        self.opcode = opcode    # Raw bytes
        self.mnemonic = mnemonic  # e.g., "mov", "jmp"
        self.operands = operands  # e.g., ["rax", "rbx"]
        self.size = len(opcode)
        self.embedding = None   # Vector embedding

    def __repr__(self):
        return f"{hex(self.address)}: {self.mnemonic} {', '.join(self.operands)}"

Function Representation

class Function:
    """Represents a function in binary."""

    def __init__(self, address, name=None):
        self.address = address
        self.name = name or f"func_{hex(address)}"
        self.instructions = []  # List of Instruction objects
        self.basic_blocks = []  # List of BasicBlock objects
        self.calls = []  # Functions this calls
        self.callers = []  # Functions that call this
        self.size = 0

    def add_instruction(self, instruction):
        """Add instruction to function."""
        self.instructions.append(instruction)
        self.size += instruction.size

    def get_control_flow_graph(self):
        """Build control flow graph."""
        cfg = {}
        for bb in self.basic_blocks:
            cfg[bb.address] = bb.successors
        return cfg

Basic Block Representation

class BasicBlock:
    """Represents a basic block (straight-line code)."""

    def __init__(self, address):
        self.address = address
        self.instructions = []
        self.successors = []  # Next basic blocks
        self.predecessors = []  # Previous basic blocks

    def is_terminator(self, instruction):
        """Check if instruction terminates block."""
        terminators = ['jmp', 'je', 'jne', 'ret', 'call']
        return instruction.mnemonic in terminators

Vector Search Algorithms

HNSW (Hierarchical Navigable Small World)

class HNSWIndex:
    """HNSW index for approximate nearest neighbor search."""

    def __init__(self, m=16, ef_construction=64, ef_search=200):
        self.m = m  # Number of connections per node
        self.ef_construction = ef_construction  # Construction parameter
        self.ef_search = ef_search  # Search parameter
        self.graph = {}  # Adjacency list
        self.data = {}  # Vector data

    def insert(self, vector_id, vector):
        """Insert vector into index."""
        # Find nearest neighbors
        neighbors = self._find_neighbors(vector, self.ef_construction)

        # Add to graph
        self.graph[vector_id] = neighbors[:self.m]
        self.data[vector_id] = vector

    def search(self, query_vector, k=10):
        """Search for k nearest neighbors."""
        # Start from random node
        candidates = self._find_neighbors(query_vector, self.ef_search)

        # Return top-k
        return sorted(
            candidates,
            key=lambda x: self._distance(query_vector, self.data[x])
        )[:k]

    def _find_neighbors(self, vector, ef):
        """Find approximate neighbors."""
        # Implementation of HNSW search algorithm
        pass

    def _distance(self, v1, v2):
        """Compute cosine distance."""
        return 1 - (np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2)))

Cosine Similarity

def cosine_similarity(v1, v2):
    """Compute cosine similarity between vectors."""
    dot_product = np.dot(v1, v2)
    norm_v1 = np.linalg.norm(v1)
    norm_v2 = np.linalg.norm(v2)

    if norm_v1 == 0 or norm_v2 == 0:
        return 0

    return dot_product / (norm_v1 * norm_v2)

def cosine_distance(v1, v2):
    """Compute cosine distance (1 - similarity)."""
    return 1 - cosine_similarity(v1, v2)

Control Flow Analysis Algorithms

Depth-First Search (DFS)

def dfs_traverse(start_block, graph):
    """Traverse control flow graph using DFS."""
    visited = set()
    stack = [start_block]
    order = []

    while stack:
        block = stack.pop()
        if block in visited:
            continue

        visited.add(block)
        order.append(block)

        # Add successors to stack
        for successor in graph.get(block, []):
            if successor not in visited:
                stack.append(successor)

    return order

Dominator Tree Construction

def compute_dominators(start_block, graph):
    """Compute dominator tree."""
    blocks = set(graph.keys())
    dominators = {block: blocks.copy() for block in blocks}
    dominators[start_block] = {start_block}

    changed = True
    while changed:
        changed = False
        for block in blocks:
            if block == start_block:
                continue

            # Intersection of dominators of predecessors
            new_dom = blocks.copy()
            for pred in get_predecessors(block, graph):
                new_dom &= dominators[pred]

            new_dom.add(block)

            if new_dom != dominators[block]:
                dominators[block] = new_dom
                changed = True

    return dominators

Embedding Generation Algorithms

Sentence Transformer Encoding

def generate_embeddings(texts, model_name='all-MiniLM-L6-v2'):
    """Generate embeddings using sentence transformers."""
    from sentence_transformers import SentenceTransformer

    model = SentenceTransformer(model_name)
    embeddings = model.encode(
        texts,
        batch_size=32,
        show_progress_bar=True,
        convert_to_numpy=True
    )

    return embeddings  # Shape: (n_texts, 384)

Batch Encoding with Caching

def batch_encode_with_cache(texts, cache, model):
    """Encode texts with caching."""
    embeddings = []
    uncached_texts = []
    uncached_indices = []

    # Check cache
    for i, text in enumerate(texts):
        text_hash = hashlib.sha256(text.encode()).hexdigest()
        cached = cache.get(f"embedding:{text_hash}")

        if cached:
            embeddings.append(cached)
        else:
            uncached_texts.append(text)
            uncached_indices.append(i)

    # Encode uncached
    if uncached_texts:
        new_embeddings = model.encode(uncached_texts, batch_size=32)

        # Cache and add to results
        for i, (text, embedding) in enumerate(zip(uncached_texts, new_embeddings)):
            text_hash = hashlib.sha256(text.encode()).hexdigest()
            cache.set(f"embedding:{text_hash}", embedding, ttl=604800)
            embeddings.insert(uncached_indices[i], embedding)

    return np.array(embeddings)

Memory Management Algorithms

Sliding Window Memory

class SlidingWindowMemory:
    """Sliding window memory with fixed size."""

    def __init__(self, window_size=3):
        self.window_size = window_size
        self.messages = []

    def add_message(self, message):
        """Add message to window."""
        self.messages.append(message)

        # Remove oldest if exceeds window size
        if len(self.messages) > self.window_size:
            self.messages.pop(0)

    def get_context(self):
        """Get current context."""
        return self.messages

Hierarchical Memory with Importance

class HierarchicalMemory:
    """Hierarchical memory with importance-based promotion."""

    def __init__(self, recent_size=3, important_size=5):
        self.recent = []
        self.important = []
        self.archived = []
        self.recent_size = recent_size
        self.important_size = important_size

    def add_message(self, message, importance=0.5):
        """Add message with importance score."""
        msg_obj = {
            'content': message,
            'importance': importance,
            'timestamp': time.time()
        }

        if importance > 0.8:
            # High importance: add to important
            self.important.append(msg_obj)
            if len(self.important) > self.important_size:
                self.archived.append(self.important.pop(0))
        else:
            # Normal: add to recent
            self.recent.append(msg_obj)
            if len(self.recent) > self.recent_size:
                self.archived.append(self.recent.pop(0))

    def get_context(self, max_messages=10):
        """Get context for LLM."""
        context = []
        context.extend(self.recent)
        context.extend(self.important)

        # Sort by timestamp (most recent first)
        context.sort(key=lambda x: x['timestamp'], reverse=True)

        return context[:max_messages]

Caching Algorithms

LRU Cache Implementation

from collections import OrderedDict

class LRUCache:
    """Least Recently Used cache."""

    def __init__(self, capacity=1000):
        self.cache = OrderedDict()
        self.capacity = capacity

    def get(self, key):
        """Get value from cache."""
        if key not in self.cache:
            return None

        # Move to end (most recently used)
        self.cache.move_to_end(key)
        return self.cache[key]

    def put(self, key, value):
        """Put value in cache."""
        if key in self.cache:
            self.cache.move_to_end(key)

        self.cache[key] = value

        # Remove least recently used if exceeds capacity
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)

Multi-Level Cache

class MultiLevelCache:
    """Multi-level cache: L1 (memory) -> L2 (Redis) -> L3 (DB)."""

    def __init__(self, l1_size=1000, l2_ttl=3600, l3_ttl=86400):
        self.l1 = LRUCache(capacity=l1_size)
        self.l2_ttl = l2_ttl
        self.l3_ttl = l3_ttl
        self.redis = redis.Redis()
        self.db = DatabaseManager()

    def get(self, key):
        """Get from cache hierarchy."""
        # Try L1
        value = self.l1.get(key)
        if value:
            return value

        # Try L2
        value = self.redis.get(key)
        if value:
            self.l1.put(key, value)
            return value

        # Try L3
        value = self.db.get(key)
        if value:
            self.redis.set(key, value, ex=self.l2_ttl)
            self.l1.put(key, value)
            return value

        return None

    def put(self, key, value):
        """Put in all cache levels."""
        self.l1.put(key, value)
        self.redis.set(key, value, ex=self.l2_ttl)
        self.db.put(key, value, ttl=self.l3_ttl)

Implementation Patterns

Agent Pattern

class BaseAgent:
    """Base class for all agents."""

    def __init__(self, name, orchestrator, api_key, model):
        self.name = name
        self.orchestrator = orchestrator
        self.api_key = api_key
        self.model = model
        self.logger = logging.getLogger(self.name)

    def execute(self, task):
        """Execute agent task."""
        try:
            self.logger.info(f"Starting {self.name}")
            result = self._execute_impl(task)
            self.logger.info(f"Completed {self.name}")
            return result
        except Exception as e:
            self.logger.exception(f"Error in {self.name}: {e}")
            raise

    def _execute_impl(self, task):
        """Implement agent logic (override in subclass)."""
        raise NotImplementedError

Pipeline Pattern

class Pipeline:
    """Execute agents in sequence."""

    def __init__(self, agents):
        self.agents = agents

    def execute(self, input_data):
        """Execute pipeline."""
        result = input_data

        for agent in self.agents:
            result = agent.execute(result)

        return result

Factory Pattern

class AgentFactory:
    """Factory for creating agents."""

    _agents = {}

    @classmethod
    def register(cls, name, agent_class):
        """Register agent class."""
        cls._agents[name] = agent_class

    @classmethod
    def create(cls, name, **kwargs):
        """Create agent instance."""
        if name not in cls._agents:
            raise ValueError(f"Unknown agent: {name}")

        return cls._agents[name](**kwargs)

# Register agents
AgentFactory.register('daa', DisassemblyAnalysisAgent)
AgentFactory.register('lima', LogicIdentificationMappingAgent)
AgentFactory.register('pea', PatchingExecutionAgent)
AgentFactory.register('va', VerificationAgent)

Testing Strategy

Unit Testing

Test Structure

# tests/unit/test_orchestrator.py
import pytest
from src.agents.orchestrator import OrchestratingAgent

class TestOrchestratingAgent:
    """Test suite for OrchestratingAgent."""

    @pytest.fixture
    def orchestrator(self):
        """Create test orchestrator."""
        return OrchestratingAgent(use_database=False)

    @pytest.fixture
    def sample_binary(self, tmp_path):
        """Create sample binary for testing."""
        binary_path = tmp_path / "test.bin"
        binary_path.write_bytes(b"\x55\x89\xe5\x83\xec\x10")  # x86 prologue
        return str(binary_path)

    def test_run_success(self, orchestrator, sample_binary):
        """Test successful binary analysis."""
        result = orchestrator.run(sample_binary)

        assert result['success']
        assert 'binary_hash' in result
        assert 'disassembly' in result

    def test_run_invalid_path(self, orchestrator):
        """Test with invalid binary path."""
        with pytest.raises(FileNotFoundError):
            orchestrator.run("/nonexistent/binary")

    def test_call_openrouter_success(self, orchestrator):
        """Test OpenRouter API call."""
        response = orchestrator.call_openrouter("Test prompt")

        assert 'content' in response
        assert response['model'] == orchestrator.model

Test Fixtures

# tests/conftest.py
import pytest
from src.utils.database import DatabaseManager
from src.utils.cache import CacheManager

@pytest.fixture(scope='session')
def test_db():
    """Create test database."""
    db = DatabaseManager(database='raverse_test')
    db.create_tables()
    yield db
    db.drop_tables()

@pytest.fixture(scope='session')
def test_cache():
    """Create test cache."""
    cache = CacheManager(redis_db=15)  # Use separate Redis DB
    yield cache
    cache.flush()

@pytest.fixture
def sample_embedding():
    """Create sample embedding."""
    return [0.1] * 384  # 384-dimensional vector

Integration Testing

End-to-End Tests

# tests/integration/test_end_to_end.py
import pytest

class TestEndToEnd:
    """End-to-end integration tests."""

    def test_complete_analysis_pipeline(self, test_db, test_cache):
        """Test complete offline pipeline."""
        from src.agents.orchestrator import OrchestratingAgent

        # Setup
        oa = OrchestratingAgent(use_database=True)
        binary_path = "tests/fixtures/test_binary.exe"

        # Execute
        result = oa.run(binary_path)

        # Verify
        assert result['success']

        # Check database
        records = test_db.execute_query(
            "SELECT * FROM raverse.binaries WHERE file_hash = %s",
            (result['binary_hash'],)
        )
        assert len(records) > 0

        # Check cache
        cached = test_cache.get_cached_analysis(result['binary_hash'])
        assert cached is not None

    def test_vector_search_integration(self, test_db):
        """Test vector search integration."""
        from src.utils.semantic_search import SemanticSearchEngine
        from src.utils.embeddings_v2 import EmbeddingGenerator

        # Setup
        embedding_gen = EmbeddingGenerator()
        search_engine = SemanticSearchEngine(test_db, None)

        # Store code
        code = "cmp eax, 0x0; je 0x401000"
        embedding = embedding_gen.generate_embedding(code)
        search_engine.store_code_embedding(
            binary_hash="test123",
            code_snippet=code,
            metadata={'function': 'main'}
        )

        # Search
        results = search_engine.find_similar_code(
            query="compare eax with zero",
            limit=5
        )

        assert len(results) > 0
        assert results[0]['similarity'] > 0.7

Performance Testing

Benchmark Tests

# tests/performance/test_benchmarks.py
import pytest
import time

class TestPerformance:
    """Performance benchmark tests."""

    @pytest.mark.benchmark
    def test_binary_analysis_performance(self, benchmark):
        """Benchmark binary analysis."""
        from src.agents.orchestrator import OrchestratingAgent

        oa = OrchestratingAgent()
        binary_path = "tests/fixtures/test_binary.exe"

        # Run benchmark
        result = benchmark(oa.run, binary_path)

        # Assert performance
        assert result['execution_time_ms'] < 5000  # < 5 seconds

    @pytest.mark.benchmark
    def test_vector_search_performance(self, benchmark):
        """Benchmark vector search."""
        from src.utils.semantic_search import SemanticSearchEngine

        search_engine = SemanticSearchEngine(None, None)
        query_embedding = [0.1] * 384

        # Run benchmark
        result = benchmark(
            search_engine.find_similar_code,
            query="test",
            limit=10
        )

        # Assert performance
        assert result['execution_time_ms'] < 100  # < 100ms

Test Coverage

Coverage Configuration

# .coveragerc
[run]
source = src
omit =
    */tests/*
    */venv/*
    setup.py

[report]
exclude_lines =
    pragma: no cover
    def __repr__
    raise AssertionError
    raise NotImplementedError
    if __name__ == .__main__.:
    if TYPE_CHECKING:
    @abstractmethod

[html]
directory = htmlcov

Running Coverage

# Generate coverage report
pytest tests/ --cov=src --cov-report=html --cov-report=term

# View report
open htmlcov/index.html

# Check coverage threshold
pytest tests/ --cov=src --cov-fail-under=80

CI/CD Pipeline

GitHub Actions Workflow

# .github/workflows/ci.yml
name: CI/CD Pipeline

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main, develop]

jobs:
  test:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: pgvector/pgvector:pg17-latest
        env:
          POSTGRES_USER: raverse
          POSTGRES_PASSWORD: test
          POSTGRES_DB: raverse_test
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 5432:5432

      redis:
        image: redis:8.2-alpine
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
        ports:
          - 6379:6379

    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.13'

      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install pytest pytest-cov pytest-xdist

      - name: Lint with ruff
        run: ruff check src/ tests/

      - name: Type check with mypy
        run: mypy src/

      - name: Run tests
        run: |
          pytest tests/ -v --cov=src --cov-report=xml
        env:
          DB_HOST: localhost
          DB_PORT: 5432
          DB_USER: raverse
          DB_PASSWORD: test
          DB_NAME: raverse_test
          REDIS_HOST: localhost
          REDIS_PORT: 6379

      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          files: ./coverage.xml
          flags: unittests
          name: codecov-umbrella

Docker Build Pipeline

# .github/workflows/docker.yml
name: Docker Build & Push

on:
  push:
    branches: [main]
    tags: ['v*']

jobs:
  build:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v2

      - name: Login to Docker Hub
        uses: docker/login-action@v2
        with:
          username: ${{ secrets.DOCKER_USERNAME }}
          password: ${{ secrets.DOCKER_PASSWORD }}

      - name: Build and push
        uses: docker/build-push-action@v4
        with:
          context: .
          push: true
          tags: |
            ${{ secrets.DOCKER_USERNAME }}/raverse:latest
            ${{ secrets.DOCKER_USERNAME }}/raverse:${{ github.sha }}
          cache-from: type=registry,ref=${{ secrets.DOCKER_USERNAME }}/raverse:buildcache
          cache-to: type=registry,ref=${{ secrets.DOCKER_USERNAME }}/raverse:buildcache,mode=max

Workflow Documentation

Offline Binary Analysis Workflow

1. INPUT: Binary file path
   ↓
2. METADATA EXTRACTION
   - File type detection (ELF, PE, Mach-O)
   - Architecture detection (x86, x64, ARM)
   - Size and hash calculation
   ↓
3. DISASSEMBLY (DAA)
   - Load binary with Capstone
   - Disassemble all code sections
   - Identify functions
   - Generate embeddings
   ↓
4. LOGIC ANALYSIS (LIMA)
   - Build control flow graph
   - Analyze data flow
   - Identify algorithms
   - LLM semantic analysis
   ↓
5. PATCH GENERATION (PEA)
   - Identify vulnerable patterns
   - Generate patches
   - Apply patches to binary
   - Create backup
   ↓
6. VERIFICATION (VA)
   - Verify binary structure
   - Verify patches applied
   - Test functionality
   - Generate report
   ↓
7. OUTPUT: Analysis result with patches

Online Analysis Workflow

1. INPUT: Target URL
   ↓
2. RECONNAISSANCE
   - Technology stack detection
   - Endpoint discovery
   - Server information gathering
   ↓
3. TRAFFIC INTERCEPTION
   - Start mitmproxy
   - Navigate application
   - Capture HTTP(S) traffic
   - Extract API calls
   ↓
4. JAVASCRIPT ANALYSIS
   - Extract JavaScript code
   - Deobfuscate
   - Analyze client-side logic
   - Extract API calls
   ↓
5. API REVERSE ENGINEERING
   - Map endpoints
   - Extract parameters
   - Detect authentication
   - Generate OpenAPI spec
   ↓
6. SECURITY ANALYSIS
   - Identify vulnerabilities
   - Generate POCs
   - Assess risk
   ↓
7. VALIDATION & REPORTING
   - Validate findings
   - Generate report
   - Export results
   ↓
8. OUTPUT: API documentation + security report

RAG Query Workflow

1. INPUT: User query
   ↓
2. EMBEDDING GENERATION
   - Convert query to embedding
   - Use sentence-transformers
   ↓
3. KNOWLEDGE RETRIEVAL
   - Search knowledge base
   - Use vector similarity
   - Filter by threshold
   - Rank by relevance
   ↓
4. CONTEXT AUGMENTATION
   - Combine query + retrieved knowledge
   - Maintain token budget
   - Format for LLM
   ↓
5. LLM GENERATION
   - Call OpenRouter API
   - Generate response
   - Include sources
   ↓
6. OUTPUT: Response with sources

Deployment Workflows

Development Deployment

# 1. Clone repository
git clone https://github.com/usemanusai/RAVERSE.git
cd RAVERSE

# 2. Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# 3. Install dependencies
pip install -r requirements.txt

# 4. Configure environment
cp .env.example .env
# Edit .env with your settings

# 5. Initialize database
python -m src.utils.database --init

# 6. Run application
python src/main.py

Production Deployment

# 1. Build Docker image
docker build -t raverse:latest .

# 2. Push to registry
docker push your-registry/raverse:latest

# 3. Deploy with Docker Compose
docker-compose -f docker-compose.prod.yml up -d

# 4. Verify deployment
docker-compose ps
docker-compose logs -f raverse

# 5. Run health checks
curl http://localhost:8000/health

Kubernetes Deployment

# 1. Create namespace
kubectl create namespace raverse

# 2. Create secrets
kubectl create secret generic raverse-secrets \
  --from-literal=api-key=$OPENROUTER_API_KEY \
  -n raverse

# 3. Deploy Helm chart
helm install raverse ./helm/raverse \
  -n raverse \
  -f helm/values-prod.yaml

# 4. Verify deployment
kubectl get pods -n raverse
kubectl logs -f deployment/raverse -n raverse

# 5. Access application
kubectl port-forward svc/raverse 8000:8000 -n raverse

Cloudflare Workflows Deployment (Hybrid-Cloud Architecture)

RAVERSE now supports deployment with Cloudflare Workflows for a hybrid-cloud architecture combining Cloudflare's edge network with Render's origin deployment.

Features:

• Binary Analysis Workflow: Single-step analysis with edge caching
• Multi-Step Analysis Workflow: DAG-based workflows with parallel execution
• Cache Management Workflow: Edge cache operations and optimization
• Hybrid Routing Workflow: Intelligent routing between edge and origin

Setup:

# 1. Navigate to workflows directory
cd workflows-starter

# 2. Install dependencies
npm install

# 3. Authenticate with Cloudflare
npx wrangler login

# 4. Setup infrastructure (KV namespaces and D1 database)
npm run setup

# 5. Set secrets
npx wrangler secret put OPENROUTER_API_KEY

# 6. Deploy to Cloudflare
npm run deploy

# 7. Verify deployment
curl https://raverse-workflows.use-manus-ai.workers.dev/health

Documentation:

• Cloudflare Workflows Setup Guide
• Cloudflare Deployment Guide
• Hybrid-Cloud Architecture Guide

Architecture:

Client → Cloudflare Workers (Edge) → Cloudflare Workflows → Render (RAVERSE API)
                    ↓
            KV Cache (RAVERSE_CACHE)
            D1 Database (raverse-workflows)

Benefits:

• Global edge caching for reduced latency
• Automatic failover and retry logic
• Durable workflow execution
• State persistence with D1 database
• Performance optimization with edge caching
• Seamless integration with existing RAVERSE deployment

Security & Compliance

Security Architecture

Defense in Depth

Layer 1: Network Security
├── TLS/SSL encryption for all connections
├── Network policies for pod-to-pod communication
├── Firewall rules for ingress/egress
└── DDoS protection

Layer 2: Application Security
├── Input validation and sanitization
├── SQL injection prevention (parameterized queries)
├── XSS protection
├── CSRF tokens
└── Rate limiting

Layer 3: Data Security
├── Encryption at rest (AES-256)
├── Encryption in transit (TLS 1.3)
├── Database encryption
├── Secrets management (Vault/K8s Secrets)
└── Data masking for sensitive fields

Layer 4: Access Control
├── Authentication (API keys, OAuth)
├── Authorization (RBAC)
├── Audit logging
├── Session management
└── Multi-factor authentication

Secrets Management

# ✓ GOOD: Secure secrets handling
import os
from dotenv import load_dotenv

# Load from .env (never commit to git)
load_dotenv()

# Get secrets from environment
api_key = os.getenv('OPENROUTER_API_KEY')
db_password = os.getenv('DB_PASSWORD')

# Validate secrets are set
if not api_key:
    raise ValueError("OPENROUTER_API_KEY not configured")

# Never log secrets
logger.info(f"Using API key: {api_key[:10]}...")  # Only show prefix

# Use Kubernetes Secrets in production
# kubectl create secret generic raverse-secrets \
#   --from-literal=api-key=$OPENROUTER_API_KEY

Input Validation

# ✓ GOOD: Comprehensive input validation
import os
from pathlib import Path

def validate_binary_path(path: str) -> str:
    """Validate binary path for security."""
    # Check path exists
    if not os.path.exists(path):
        raise FileNotFoundError(f"Binary not found: {path}")

    # Check path is file
    if not os.path.isfile(path):
        raise ValueError(f"Path is not a file: {path}")

    # Check path is within allowed directory
    allowed_dir = os.path.abspath('/binaries')
    real_path = os.path.abspath(path)

    if not real_path.startswith(allowed_dir):
        raise ValueError(f"Path outside allowed directory: {path}")

    # Check file size (prevent DoS)
    max_size = 1024 * 1024 * 100  # 100 MB
    if os.path.getsize(real_path) > max_size:
        raise ValueError(f"Binary too large: {os.path.getsize(real_path)} bytes")

    return real_path

SQL Injection Prevention

# ✓ GOOD: Parameterized queries
from src.utils.database import DatabaseManager

db = DatabaseManager()

# GOOD: Parameterized query
result = db.execute_query(
    "SELECT * FROM binaries WHERE file_hash = %s",
    (user_input,)  # Parameters passed separately
)

# BAD: String concatenation (vulnerable)
# result = db.execute_query(f"SELECT * FROM binaries WHERE file_hash = '{user_input}'")

Compliance

GDPR Compliance

# Data retention policy
DATA_RETENTION_POLICY = {
    'analysis_results': 90,  # days
    'user_data': 365,
    'logs': 30,
    'backups': 90
}

# Right to be forgotten
def delete_user_data(user_id):
    """Delete all user data (GDPR right to be forgotten)."""
    db = DatabaseManager()

    # Delete analysis results
    db.execute_query(
        "DELETE FROM analysis_results WHERE user_id = %s",
        (user_id,)
    )

    # Delete user record
    db.execute_query(
        "DELETE FROM users WHERE id = %s",
        (user_id,)
    )

    # Delete from cache
    cache = CacheManager()
    cache.delete_user_cache(user_id)

    # Log deletion
    logger.info(f"User data deleted: {user_id}")

HIPAA Compliance (if handling health data)

# Audit logging for HIPAA
def log_access(user_id, resource_id, action):
    """Log access for audit trail."""
    audit_log = {
        'timestamp': datetime.utcnow(),
        'user_id': user_id,
        'resource_id': resource_id,
        'action': action,
        'ip_address': get_client_ip(),
        'user_agent': get_user_agent()
    }

    db = DatabaseManager()
    db.execute_query(
        "INSERT INTO audit_log (timestamp, user_id, resource_id, action, ip_address, user_agent) "
        "VALUES (%s, %s, %s, %s, %s, %s)",
        (audit_log['timestamp'], audit_log['user_id'], audit_log['resource_id'],
         audit_log['action'], audit_log['ip_address'], audit_log['user_agent'])
    )

SOC 2 Compliance

# SOC 2 requirements
SOC2_REQUIREMENTS = {
    'CC6.1': 'Logical access controls',
    'CC6.2': 'Prior to issuing system credentials',
    'CC7.1': 'System monitoring and alerting',
    'CC7.2': 'System monitoring tools',
    'CC7.3': 'Unauthorized activities detection',
    'CC7.4': 'Identified security incidents response',
    'CC8.1': 'Incident response procedures',
    'CC9.1': 'Change management procedures'
}

# Implement monitoring
def setup_monitoring():
    """Setup SOC 2 monitoring."""
    # Enable audit logging
    enable_audit_logging()

    # Setup alerting
    setup_alerts()

    # Enable encryption
    enable_encryption()

    # Setup access controls
    setup_rbac()

Feature Documentation

Binary Analysis Features

Vulnerability Detection

# Detect common vulnerability patterns
VULNERABILITY_PATTERNS = {
    'buffer_overflow': {
        'pattern': r'mov.*\[.*\+.*\]',
        'description': 'Potential buffer overflow',
        'severity': 'high'
    },
    'use_after_free': {
        'pattern': r'mov.*\[.*\].*free',
        'description': 'Potential use-after-free',
        'severity': 'high'
    },
    'integer_overflow': {
        'pattern': r'add.*jno',
        'description': 'Potential integer overflow',
        'severity': 'medium'
    },
    'format_string': {
        'pattern': r'printf.*%x',
        'description': 'Potential format string vulnerability',
        'severity': 'high'
    }
}

def detect_vulnerabilities(disassembly):
    """Detect vulnerabilities in disassembly."""
    vulnerabilities = []

    for vuln_name, vuln_info in VULNERABILITY_PATTERNS.items():
        pattern = vuln_info['pattern']

        for instruction in disassembly:
            if re.match(pattern, instruction):
                vulnerabilities.append({
                    'type': vuln_name,
                    'description': vuln_info['description'],
                    'severity': vuln_info['severity'],
                    'instruction': instruction
                })

    return vulnerabilities

Patch Generation

# Patch generation strategies
PATCH_STRATEGIES = {
    'nop_padding': {
        'description': 'Replace vulnerable code with NOPs',
        'risk': 'low',
        'effectiveness': 'medium'
    },
    'bounds_check': {
        'description': 'Add bounds checking before access',
        'risk': 'low',
        'effectiveness': 'high'
    },
    'return_early': {
        'description': 'Add early return to skip vulnerable code',
        'risk': 'medium',
        'effectiveness': 'high'
    },
    'exception_handler': {
        'description': 'Wrap in exception handler',
        'risk': 'medium',
        'effectiveness': 'medium'
    }
}

def generate_patches(vulnerabilities):
    """Generate patches for vulnerabilities."""
    patches = []

    for vuln in vulnerabilities:
        if vuln['severity'] == 'high':
            strategy = 'bounds_check'
        elif vuln['severity'] == 'medium':
            strategy = 'return_early'
        else:
            strategy = 'nop_padding'

        patch = {
            'vulnerability': vuln,
            'strategy': strategy,
            'description': PATCH_STRATEGIES[strategy]['description'],
            'risk': PATCH_STRATEGIES[strategy]['risk']
        }

        patches.append(patch)

    return patches

Online Analysis Features

Technology Detection

# Technology stack detection
TECHNOLOGY_SIGNATURES = {
    'frameworks': {
        'Django': ['django', 'csrf_token', 'django.core'],
        'Flask': ['flask', 'werkzeug', 'jinja2'],
        'React': ['react', 'react-dom', '__REACT_DEVTOOLS_GLOBAL_HOOK__'],
        'Vue': ['vue', '__VUE__', 'Vue.js'],
        'Angular': ['angular', 'ng-app', 'ng-controller']
    },
    'databases': {
        'PostgreSQL': ['psycopg2', 'pg_', 'postgres'],
        'MySQL': ['mysql', 'mysqli', 'PDO'],
        'MongoDB': ['mongodb', 'mongoose', 'mongo'],
        'Redis': ['redis', 'ioredis', 'redis-py']
    },
    'servers': {
        'Apache': ['Apache', 'mod_', 'httpd'],
        'Nginx': ['nginx', 'Nginx'],
        'IIS': ['IIS', 'ASP.NET'],
        'Node.js': ['Node.js', 'Express', 'npm']
    }
}

def detect_technologies(html_content, headers, js_code):
    """Detect technologies in web application."""
    detected = {
        'frameworks': [],
        'databases': [],
        'servers': []
    }

    content = html_content + js_code + str(headers)

    for category, signatures in TECHNOLOGY_SIGNATURES.items():
        for tech, patterns in signatures.items():
            for pattern in patterns:
                if pattern.lower() in content.lower():
                    detected[category].append(tech)

    return detected

API Endpoint Discovery

# API endpoint patterns
API_PATTERNS = {
    'rest': r'/api/v\d+/[a-z_/]+',
    'graphql': r'/graphql',
    'websocket': r'wss?://',
    'rpc': r'/rpc',
    'soap': r'\.wsdl$'
}

def discover_api_endpoints(traffic_data):
    """Discover API endpoints from traffic."""
    endpoints = []

    for request in traffic_data:
        url = request['url']
        method = request['method']

        for api_type, pattern in API_PATTERNS.items():
            if re.match(pattern, url):
                endpoints.append({
                    'url': url,
                    'method': method,
                    'type': api_type,
                    'parameters': extract_parameters(request),
                    'authentication': detect_authentication(request)
                })

    return endpoints

Memory Management Features

Hierarchical Memory

# Hierarchical memory with importance-based promotion
class HierarchicalMemoryAgent:
    """Agent with hierarchical memory."""

    def __init__(self):
        self.recent = []  # Recent messages (window size: 3)
        self.important = []  # Important messages (size: 5)
        self.archived = []  # Archived messages (unlimited)

    def add_message(self, message, importance=0.5):
        """Add message with importance score."""
        msg = {
            'content': message,
            'importance': importance,
            'timestamp': time.time()
        }

        if importance > 0.8:
            # High importance: promote to important
            self.important.append(msg)
            if len(self.important) > 5:
                self.archived.append(self.important.pop(0))
        else:
            # Normal: add to recent
            self.recent.append(msg)
            if len(self.recent) > 3:
                self.archived.append(self.recent.pop(0))

    def get_context(self, max_messages=10):
        """Get context for LLM."""
        context = []
        context.extend(self.recent)
        context.extend(self.important)

        # Sort by timestamp (most recent first)
        context.sort(key=lambda x: x['timestamp'], reverse=True)

        return context[:max_messages]

Retrieval-Based Memory (RAG)

# Retrieval-based memory with semantic search
class RetrievalMemoryAgent:
    """Agent with retrieval-based memory."""

    def __init__(self, db, embedding_gen):
        self.db = db
        self.embedding_gen = embedding_gen

    def store_memory(self, content, metadata=None):
        """Store memory with embedding."""
        embedding = self.embedding_gen.generate_embedding(content)

        self.db.execute_query(
            "INSERT INTO memory (content, embedding, metadata) VALUES (%s, %s, %s)",
            (content, embedding, json.dumps(metadata or {}))
        )

    def retrieve_memory(self, query, limit=5, threshold=0.7):
        """Retrieve relevant memories."""
        query_embedding = self.embedding_gen.generate_embedding(query)

        results = self.db.search_similar_instructions(
            embedding=query_embedding,
            limit=limit,
            threshold=threshold
        )

        return results

    def get_context(self, query):
        """Get context for LLM based on query."""
        memories = self.retrieve_memory(query)

        context = "\n".join([
            f"- {m['content']} (relevance: {m['similarity']:.2%})"
            for m in memories
        ])

        return context

RAG Features

Knowledge Base Management

# Knowledge base with semantic search
class KnowledgeBase:
    """Semantic knowledge base."""

    def __init__(self, db, embedding_gen):
        self.db = db
        self.embedding_gen = embedding_gen

    def add_knowledge(self, content, category, tags=None):
        """Add knowledge to base."""
        embedding = self.embedding_gen.generate_embedding(content)

        self.db.execute_query(
            "INSERT INTO knowledge_base (content, category, tags, embedding) "
            "VALUES (%s, %s, %s, %s)",
            (content, category, json.dumps(tags or []), embedding)
        )

    def search(self, query, category=None, limit=10):
        """Search knowledge base."""
        query_embedding = self.embedding_gen.generate_embedding(query)

        sql = """
            SELECT content, category, tags,
                   1 - (embedding <=> %s::vector) AS similarity
            FROM knowledge_base
            WHERE 1 - (embedding <=> %s::vector) >= 0.7
        """
        params = [query_embedding, query_embedding]

        if category:
            sql += " AND category = %s"
            params.append(category)

        sql += " ORDER BY embedding <=> %s::vector LIMIT %s"
        params.extend([query_embedding, limit])

        return self.db.execute_query(sql, tuple(params))

    def get_statistics(self):
        """Get knowledge base statistics."""
        stats = self.db.execute_query(
            "SELECT category, COUNT(*) as count FROM knowledge_base GROUP BY category"
        )

        return {s['category']: s['count'] for s in stats}

Advanced Features

Batch Processing

Batch Binary Analysis

# Analyze multiple binaries efficiently
def batch_analyze_binaries(binary_paths, batch_size=5):
    """Analyze multiple binaries with batching."""
    from concurrent.futures import ThreadPoolExecutor

    oa = OrchestratingAgent()
    results = []

    with ThreadPoolExecutor(max_workers=batch_size) as executor:
        futures = [
            executor.submit(oa.run, path)
            for path in binary_paths
        ]

        for future in futures:
            try:
                result = future.result(timeout=300)
                results.append(result)
            except Exception as e:
                logger.error(f"Analysis failed: {e}")
                results.append({'error': str(e)})

    return results

# Usage
binaries = [
    '/binaries/app1.exe',
    '/binaries/app2.exe',
    '/binaries/app3.exe'
]

results = batch_analyze_binaries(binaries, batch_size=3)
for result in results:
    print(f"Binary: {result.get('binary_hash', 'ERROR')}")
    print(f"Success: {result.get('success', False)}")

Batch Embedding Generation

# Generate embeddings for large datasets
def batch_generate_embeddings(texts, batch_size=32, cache=None):
    """Generate embeddings with caching."""
    from src.utils.embeddings_v2 import EmbeddingGenerator

    embedding_gen = EmbeddingGenerator(batch_size=batch_size)
    embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]

        # Check cache
        if cache:
            batch_embeddings = []
            uncached = []
            uncached_indices = []

            for j, text in enumerate(batch):
                cached = cache.get(f"embedding:{hash(text)}")
                if cached:
                    batch_embeddings.append(cached)
                else:
                    uncached.append(text)
                    uncached_indices.append(j)

            # Generate uncached
            if uncached:
                new_embeddings = embedding_gen.batch_encode(uncached)
                for text, embedding in zip(uncached, new_embeddings):
                    cache.set(f"embedding:{hash(text)}", embedding)
                    batch_embeddings.insert(uncached_indices[len(batch_embeddings)], embedding)
        else:
            batch_embeddings = embedding_gen.batch_encode(batch)

        embeddings.extend(batch_embeddings)
        logger.info(f"Generated {len(embeddings)}/{len(texts)} embeddings")

    return embeddings

Streaming & Async Processing

Async Agent Execution

# Execute agents asynchronously
import asyncio

async def execute_agents_async(agents, task):
    """Execute agents concurrently."""
    tasks = [
        asyncio.create_task(agent.execute_async(task))
        for agent in agents
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

# Usage
async def main():
    agents = [
        ReconnaissanceAgent(orchestrator),
        TrafficInterceptionAgent(orchestrator),
        JavaScriptAnalysisAgent(orchestrator)
    ]

    task = {'target_url': 'https://api.example.com'}
    results = await execute_agents_async(agents, task)

    for agent, result in zip(agents, results):
        print(f"{agent.name}: {result}")

asyncio.run(main())

Streaming Results

# Stream results as they become available
def stream_analysis_results(binary_paths):
    """Stream analysis results."""
    oa = OrchestratingAgent()

    for binary_path in binary_paths:
        try:
            result = oa.run(binary_path)
            yield {
                'status': 'success',
                'binary_path': binary_path,
                'result': result
            }
        except Exception as e:
            yield {
                'status': 'error',
                'binary_path': binary_path,
                'error': str(e)
            }

# Usage
for result in stream_analysis_results(binary_paths):
    if result['status'] == 'success':
        print(f"✓ {result['binary_path']}")
    else:
        print(f"✗ {result['binary_path']}: {result['error']}")

Advanced Caching Strategies

Distributed Caching

# Distributed cache with Redis cluster
class DistributedCache:
    """Distributed cache using Redis cluster."""

    def __init__(self, nodes):
        from rediscluster import RedisCluster

        self.cluster = RedisCluster(
            startup_nodes=nodes,
            skip_full_coverage_check=True
        )

    def get(self, key):
        """Get from distributed cache."""
        value = self.cluster.get(key)
        return json.loads(value) if value else None

    def set(self, key, value, ttl=3600):
        """Set in distributed cache."""
        self.cluster.setex(
            key,
            ttl,
            json.dumps(value)
        )

    def delete(self, key):
        """Delete from distributed cache."""
        self.cluster.delete(key)

    def flush(self):
        """Flush all cache."""
        self.cluster.flushall()

# Usage
nodes = [
    {'host': 'redis-1', 'port': 6379},
    {'host': 'redis-2', 'port': 6379},
    {'host': 'redis-3', 'port': 6379}
]

cache = DistributedCache(nodes)
cache.set('key', {'data': 'value'})
result = cache.get('key')

Cache Warming

# Pre-populate cache with frequently accessed data
def warm_cache(cache, db):
    """Warm cache with frequently accessed data."""

    # Get frequently analyzed binaries
    frequent_binaries = db.execute_query("""
        SELECT file_hash, analysis_result
        FROM analysis_results
        WHERE created_at > NOW() - INTERVAL '7 days'
        ORDER BY access_count DESC
        LIMIT 1000
    """)

    for binary in frequent_binaries:
        cache.set(
            f"analysis:{binary['file_hash']}",
            binary['analysis_result'],
            ttl=604800  # 7 days
        )

    logger.info(f"Warmed cache with {len(frequent_binaries)} entries")

Advanced Monitoring

Custom Metrics

# Define custom metrics
from prometheus_client import Counter, Histogram, Gauge

# Counters
binary_analysis_total = Counter(
    'binary_analysis_total',
    'Total binary analyses',
    ['status', 'architecture']
)

vulnerability_detected_total = Counter(
    'vulnerability_detected_total',
    'Total vulnerabilities detected',
    ['type', 'severity']
)

# Histograms
analysis_duration_seconds = Histogram(
    'analysis_duration_seconds',
    'Analysis duration',
    buckets=(1, 2, 5, 10, 30, 60, 120)
)

patch_size_bytes = Histogram(
    'patch_size_bytes',
    'Patch size in bytes',
    buckets=(10, 50, 100, 500, 1000, 5000)
)

# Gauges
active_analyses = Gauge(
    'active_analyses',
    'Number of active analyses'
)

cache_size_bytes = Gauge(
    'cache_size_bytes',
    'Cache size in bytes'
)

# Usage
@active_analyses.track_inprogress()
def analyze_binary(binary_path):
    """Analyze binary with metrics."""
    with analysis_duration_seconds.time():
        result = oa.run(binary_path)

    binary_analysis_total.labels(
        status='success' if result['success'] else 'failed',
        architecture=result['metadata']['architecture']
    ).inc()

    for vuln in result.get('vulnerabilities', []):
        vulnerability_detected_total.labels(
            type=vuln['type'],
            severity=vuln['severity']
        ).inc()

    return result

Alerting Rules

# prometheus-alerts.yml
groups:
  - name: raverse_alerts
    rules:
      # High error rate
      - alert: HighAnalysisErrorRate
        expr: rate(binary_analysis_total{status="failed"}[5m]) > 0.1
        for: 5m
        annotations:
          summary: "High binary analysis error rate"
          description: "Error rate is {{ $value | humanizePercentage }}"

      # Slow analysis
      - alert: SlowAnalysis
        expr: histogram_quantile(0.95, analysis_duration_seconds) > 30
        for: 10m
        annotations:
          summary: "Analysis taking too long"
          description: "p95 latency is {{ $value }}s"

      # Cache efficiency
      - alert: LowCacheHitRatio
        expr: cache_hit_ratio < 0.5
        for: 15m
        annotations:
          summary: "Low cache hit ratio"
          description: "Cache hit ratio is {{ $value | humanizePercentage }}"

      # Database connection pool
      - alert: DatabaseConnectionPoolExhausted
        expr: database_connection_pool_size >= 20
        for: 2m
        annotations:
          summary: "Database connection pool exhausted"
          description: "Active connections: {{ $value }}"

Advanced Query Optimization

Query Plan Analysis

# Analyze and optimize queries
def analyze_query_performance(db, query):
    """Analyze query performance."""

    # Get query plan
    plan = db.execute_query(f"EXPLAIN ANALYZE {query}")

    # Extract metrics
    metrics = {
        'total_cost': None,
        'rows': None,
        'execution_time': None,
        'planning_time': None
    }

    for row in plan:
        if 'Total Cost' in row:
            metrics['total_cost'] = float(row.split(':')[1])
        elif 'Rows' in row:
            metrics['rows'] = int(row.split(':')[1])
        elif 'Execution Time' in row:
            metrics['execution_time'] = float(row.split(':')[1])
        elif 'Planning Time' in row:
            metrics['planning_time'] = float(row.split(':')[1])

    return metrics

# Usage
query = """
    SELECT * FROM code_embeddings
    WHERE 1 - (embedding <=> query_embedding::vector) >= 0.7
    ORDER BY embedding <=> query_embedding::vector
    LIMIT 10
"""

metrics = analyze_query_performance(db, query)
print(f"Total Cost: {metrics['total_cost']}")
print(f"Execution Time: {metrics['execution_time']}ms")

Index Optimization

# Optimize indexes
def optimize_indexes(db):
    """Optimize database indexes."""

    # Analyze index usage
    index_stats = db.execute_query("""
        SELECT schemaname, tablename, indexname, idx_scan, idx_tup_read, idx_tup_fetch
        FROM pg_stat_user_indexes
        ORDER BY idx_scan DESC
    """)

    # Identify unused indexes
    unused_indexes = [
        idx for idx in index_stats
        if idx['idx_scan'] == 0
    ]

    # Identify inefficient indexes
    inefficient_indexes = [
        idx for idx in index_stats
        if idx['idx_tup_read'] > 0 and idx['idx_tup_fetch'] / idx['idx_tup_read'] < 0.1
    ]

    logger.info(f"Unused indexes: {len(unused_indexes)}")
    logger.info(f"Inefficient indexes: {len(inefficient_indexes)}")

    # Reindex
    for idx in inefficient_indexes:
        db.execute_query(f"REINDEX INDEX {idx['indexname']}")
        logger.info(f"Reindexed: {idx['indexname']}")

Detailed Examples

Example 1: Complete Binary Analysis Workflow

# Complete workflow from binary to patched binary
from src.agents.orchestrator import OrchestratingAgent
from src.utils.database import DatabaseManager
from src.utils.cache import CacheManager

# Initialize
oa = OrchestratingAgent()
db = DatabaseManager()
cache = CacheManager()

# 1. Analyze binary
binary_path = '/binaries/vulnerable_app.exe'
result = oa.run(binary_path)

# 2. Check results
if result['success']:
    print(f"✓ Analysis successful")
    print(f"  Binary Hash: {result['binary_hash']}")
    print(f"  Vulnerabilities: {len(result.get('vulnerabilities', []))}")
    print(f"  Patches Generated: {len(result.get('patches', []))}")

    # 3. Review patches
    for patch in result['patches']:
        print(f"\n  Patch: {patch['type']}")
        print(f"  Address: {hex(patch['address'])}")
        print(f"  Risk: {patch['risk']}")

    # 4. Apply patches
    patched_binary = result['patched_binary_path']
    print(f"\n✓ Patched binary: {patched_binary}")

    # 5. Verify patches
    verification = result['verification']
    print(f"\n  Verification:")
    print(f"  - Structure Valid: {verification['structure_valid']}")
    print(f"  - Patches Applied: {verification['patch_applied']}")
    print(f"  - Functionality OK: {verification['functionality_ok']}")
else:
    print(f"✗ Analysis failed: {result.get('error')}")

Example 2: Online API Discovery

# Discover and document APIs
from src.agents.online_orchestrator import OnlineOrchestrationAgent

# Initialize
oa = OnlineOrchestrationAgent(
    api_key='sk-or-v1-...',
    model='meta-llama/llama-3.3-70b-instruct:free'
)

# Execute online analysis
result = oa.execute(
    target_url='https://api.example.com',
    scope={
        'target_url': 'https://api.example.com',
        'allowed_domains': ['api.example.com', '*.example.com'],
        'max_depth': 3,
        'max_urls': 1000
    },
    options={
        'recon': {'detect_technologies': True},
        'traffic': {'duration_seconds': 60},
        'api_discovery': {'detect_rest': True, 'detect_graphql': True},
        'security': {'check_vulnerabilities': True}
    }
)

# Process results
print("=== API Discovery Results ===\n")

# Technologies
print("Technologies Detected:")
for tech in result['recon']['technologies']:
    print(f"  - {tech}")

# API Endpoints
print("\nAPI Endpoints:")
for endpoint in result['api_reeng']['endpoints']:
    print(f"  {endpoint['method']} {endpoint['url']}")
    if endpoint['parameters']:
        for param in endpoint['parameters']:
            print(f"    - {param['name']}: {param['type']}")

# Vulnerabilities
print("\nVulnerabilities:")
for vuln in result['security']['vulnerabilities']:
    print(f"  [{vuln['severity']}] {vuln['type']}")
    print(f"    {vuln['description']}")

# Export OpenAPI spec
openapi_spec = result['api_reeng']['openapi_spec']
with open('api_spec.json', 'w') as f:
    json.dump(openapi_spec, f, indent=2)
print("\n✓ OpenAPI spec exported to api_spec.json")

Example 3: RAG-Enhanced Analysis

# Use RAG for intelligent analysis
from src.agents.online_rag_orchestrator_agent import RAGOrchestratorAgent
from src.utils.semantic_search import SemanticSearchEngine

# Initialize
rag = RAGOrchestratorAgent(
    api_key='sk-or-v1-...',
    model='meta-llama/llama-3.3-70b-instruct:free'
)

# Execute RAG query
result = rag.execute({
    'query': 'What are the common binary patching techniques?',
    'context': 'Binary analysis and security patching'
})

# Process results
print("=== RAG Analysis Results ===\n")

print("Retrieved Knowledge:")
for knowledge in result['retrieved_knowledge']:
    print(f"  - {knowledge['content']}")
    print(f"    Relevance: {knowledge['similarity']:.2%}\n")

print("Generated Response:")
print(result['generated_response'])

print(f"\nConfidence: {result['confidence']:.2%}")

Frequently Asked Questions (FAQ)

General Questions

Q: What is RAVERSE 2.0? A: RAVERSE 2.0 is an advanced AI-powered multi-agent system for binary analysis, reverse engineering, and automated patching. It combines offline binary patching with online target analysis using 35+ specialized AI agents.

Q: What are the system requirements? A: Minimum: 2 CPU cores, 4GB RAM, 20GB disk. Recommended: 8 CPU cores, 16GB RAM, 100GB SSD. Requires Python 3.13+, PostgreSQL 17, Redis 8.2.

Q: Can I use RAVERSE for commercial purposes? A: Yes, RAVERSE is licensed under MIT License, allowing commercial use. However, you must only analyze binaries and systems you own or are authorized to analyze.

Q: How accurate is the vulnerability detection? A: Accuracy depends on binary complexity and LLM model used. Typical accuracy: 85-95% for common vulnerabilities. Always verify findings manually.

Q: What LLM models are supported? A: RAVERSE uses OpenRouter API, supporting 100+ models including GPT-4, Claude, Llama, Mistral, etc. Free models available for testing.

Technical Questions

Q: How does vector search work? A: RAVERSE uses pgvector with HNSW indexing for approximate nearest neighbor search. Embeddings are 384-dimensional vectors generated using sentence-transformers. Cosine distance is used for similarity.

Q: What is the maximum binary size? A: Recommended maximum: 100MB. Larger binaries may require more memory and time. Can be split into chunks for analysis.

Q: How long does analysis take? A: Typical times: Small binary (<1MB): 2-3s, Medium (1-10MB): 5-10s, Large (10-100MB): 30-60s. Depends on binary complexity and LLM model.

Q: Can I use RAVERSE offline? A: Offline binary analysis works without internet. Online analysis requires internet for LLM API calls and web reconnaissance.

Q: How is data stored? A: Analysis results stored in PostgreSQL with pgvector. Cache in Redis. Logs in files. All data encrypted at rest and in transit.

Deployment Questions

Q: How do I deploy RAVERSE to production? A: Use Docker Compose for simple deployments or Kubernetes for scalable deployments. See Production Deployment Guide section.

Q: Can I scale RAVERSE horizontally? A: Yes, deploy multiple agent worker pods behind a load balancer. Scale PostgreSQL with read replicas and Redis with cluster mode.

Q: What monitoring tools are recommended? A: Prometheus for metrics collection, Grafana for visualization. RAVERSE exposes Prometheus metrics by default.

Q: How do I backup data? A: Use pg_dump for PostgreSQL backups. Automate with cron jobs. Store backups in S3 or other object storage.

Troubleshooting Questions

Q: Analysis is very slow. What can I do? A: 1) Use faster LLM model, 2) Reduce binary size, 3) Increase CPU/RAM, 4) Check database performance, 5) Enable caching.

Q: Getting "API rate limit exceeded" errors? A: Implement rate limiting in your code. Use queue for requests. Consider upgrading to paid OpenRouter plan.

Q: Vector search returns no results? A: 1) Check embeddings are generated, 2) Lower similarity threshold, 3) Verify index exists, 4) Check query embedding.

Q: Database connection pool exhausted? A: Increase pool size in configuration. Check for connection leaks. Monitor active connections.

Q: Memory usage is very high? A: 1) Reduce memory preset, 2) Reduce batch size, 3) Clear cache, 4) Reduce embedding cache size.

Technical Deep Dives

Binary Format Support

Supported Formats

Format	Architecture	Status	Notes
ELF	x86, x64, ARM, ARM64	✓ Full	Linux binaries
PE	x86, x64	✓ Full	Windows binaries
Mach-O	x64, ARM64	✓ Full	macOS binaries
WebAssembly	WASM	✓ Full	Web binaries
Java	JVM	⚠ Partial	Requires decompilation

Format Detection

def detect_binary_format(binary_path):
    """Detect binary format."""
    with open(binary_path, 'rb') as f:
        magic = f.read(4)

    if magic == b'\x7fELF':
        return 'ELF'
    elif magic == b'MZ':
        return 'PE'
    elif magic == b'\xfe\xed\xfa':
        return 'Mach-O'
    elif magic == b'\x00\x61\x73\x6d':
        return 'WebAssembly'
    else:
        return 'Unknown'

Instruction Set Support

Supported ISAs

ISA	Bits	Status	Notes
x86	32	✓ Full	Intel/AMD 32-bit
x86-64	64	✓ Full	Intel/AMD 64-bit
ARM	32	✓ Full	ARM 32-bit
ARM64	64	✓ Full	ARM 64-bit
MIPS	32/64	⚠ Partial	Limited support
PowerPC	32/64	⚠ Partial	Limited support

Embedding Model Comparison

Model	Dimensions	Speed	Quality	Use Case
all-MiniLM-L6-v2	384	Very Fast	Good	General purpose
all-mpnet-base-v2	768	Medium	Excellent	High accuracy
all-MiniLM-L12-v2	384	Fast	Good	Lightweight
all-roberta-large-v1	1024	Slow	Excellent	Maximum quality

LLM Model Comparison

Model	Speed	Quality	Cost	Best For
Llama 3.3 70B	Fast	Good	Free	Testing
Mistral 7B	Very Fast	Fair	Free	Quick analysis
GPT-4 Turbo	Medium	Excellent	High	Production
Claude 3 Opus	Medium	Excellent	High	Complex analysis

Database Schema Details

Core Tables

-- Binaries table
CREATE TABLE raverse.binaries (
    id SERIAL PRIMARY KEY,
    file_name VARCHAR(255),
    file_path TEXT,
    file_hash VARCHAR(64) UNIQUE,
    file_size BIGINT,
    file_type VARCHAR(50),
    architecture VARCHAR(50),
    metadata JSONB,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- Disassembly cache
CREATE TABLE raverse.disassembly_cache (
    id SERIAL PRIMARY KEY,
    binary_id INTEGER REFERENCES raverse.binaries(id),
    address BIGINT,
    instruction VARCHAR(255),
    opcode BYTEA,
    operands TEXT,
    disassembly_text TEXT,
    embedding vector(384),
    created_at TIMESTAMP DEFAULT NOW()
);

-- Code embeddings
CREATE TABLE raverse.code_embeddings (
    id SERIAL PRIMARY KEY,
    binary_hash VARCHAR(64),
    code_snippet TEXT,
    embedding vector(384),
    metadata JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Analysis results
CREATE TABLE raverse.analysis_results (
    id SERIAL PRIMARY KEY,
    binary_id INTEGER REFERENCES raverse.binaries(id),
    result_data JSONB,
    vulnerabilities JSONB,
    patches JSONB,
    verification JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

-- Knowledge base
CREATE TABLE raverse.knowledge_base (
    id SERIAL PRIMARY KEY,
    content TEXT,
    category VARCHAR(100),
    tags TEXT[],
    embedding vector(384),
    metadata JSONB,
    created_at TIMESTAMP DEFAULT NOW()
);

Performance Tuning Parameters

PostgreSQL Tuning

-- Connection settings
max_connections = 200
superuser_reserved_connections = 10

-- Memory settings
shared_buffers = 4GB
effective_cache_size = 12GB
work_mem = 20MB
maintenance_work_mem = 1GB

-- WAL settings
wal_buffers = 16MB
checkpoint_completion_target = 0.9
wal_level = replica

-- Query planning
random_page_cost = 1.1
effective_io_concurrency = 200

-- HNSW index settings
hnsw.ef_construction = 128
hnsw.ef_search = 200

Redis Tuning

# Memory settings
maxmemory 2gb
maxmemory-policy allkeys-lru

# Persistence
save 900 1
save 300 10
save 60 10000

# Replication
repl-diskless-sync yes
repl-diskless-sync-delay 5

# Cluster settings (if using cluster)
cluster-enabled yes
cluster-node-timeout 15000

Network Configuration

Firewall Rules

# Ingress rules
Allow 8000/tcp from 0.0.0.0/0  # Application
Allow 5432/tcp from 10.0.0.0/8  # PostgreSQL (internal only)
Allow 6379/tcp from 10.0.0.0/8  # Redis (internal only)
Allow 9090/tcp from 10.0.0.0/8  # Prometheus (internal only)
Allow 3000/tcp from 10.0.0.0/8  # Grafana (internal only)

# Egress rules
Allow 443/tcp to 0.0.0.0/0  # HTTPS (OpenRouter API)
Allow 53/udp to 0.0.0.0/0   # DNS

Load Balancer Configuration

# Nginx configuration
upstream raverse_backend {
    least_conn;
    server raverse-1:8000 weight=1;
    server raverse-2:8000 weight=1;
    server raverse-3:8000 weight=1;
}

server {
    listen 80;
    server_name api.raverse.example.com;

    location / {
        proxy_pass http://raverse_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # Timeouts
        proxy_connect_timeout 60s;
        proxy_send_timeout 60s;
        proxy_read_timeout 60s;
    }
}

Logging Configuration

Log Levels

# DEBUG: Detailed execution flow
logger.debug(f"Processing instruction: {instruction}")

# INFO: Major milestones
logger.info(f"Starting analysis of {binary_path}")

# WARNING: Recoverable errors
logger.warning(f"Patch verification failed, using fallback")

# ERROR: Unrecoverable errors
logger.error(f"Analysis failed: {error}")

# CRITICAL: System-level failures
logger.critical(f"Database connection lost")

Structured Logging

# JSON structured logging
import json

log_entry = {
    'timestamp': datetime.utcnow().isoformat(),
    'level': 'INFO',
    'logger': 'orchestrator',
    'message': 'Analysis complete',
    'binary_id': 123,
    'execution_time_ms': 5432,
    'success': True,
    'tags': ['offline', 'analysis']
}

logger.info(json.dumps(log_entry))

Version History

Version 2.0.0 (Current)

• Multi-agent architecture with 35+ agents
• Offline bina