bugbounty_osint_workflow

Instructions

Implementation Reference

• src/mcp_server/app.py:1403-1423 (handler)

  MCP tool handler function that implements the bugbounty_osint_workflow tool by proxying the request to the REST API backend.

  def bugbounty_osint_workflow(domain: str) -> dict[str, Any]:
      """Create OSINT gathering workflow for bug bounty hunting.
  
      Args:
          domain: Target domain
  
      Returns:
          OSINT gathering workflow
      """
      data = {"domain": domain}
  
      logger.info(f"🕵️ Creating OSINT workflow for {domain}")
      result = api_client.safe_post("api/bugbounty/osint-workflow", data)
  
      if result.get("success"):
          logger.info(f"✅ OSINT workflow created for {domain}")
      else:
          logger.error(f"❌ Failed to create OSINT workflow for {domain}")
  
      return result

• src/rest_api_server/workflows/osint/osint.py:12-33 (handler)

  Backend REST API handler for the /api/bugbounty/osint-workflow endpoint that processes the request and delegates to the workflow manager.

  @workflow()
  def create_osint_workflow():
      """Create OSINT gathering workflow."""
      # Import here to avoid circular imports
      from src.rest_api_server.managers import BugBountyTarget, bugbounty_manager
  
      data = request.get_json()
  
      domain = data["domain"]
  
      logger.info(f"Creating OSINT workflow for {domain}")
  
      # Create bug bounty target
      target = BugBountyTarget(domain=domain)
  
      # Generate OSINT workflow
      workflow = bugbounty_manager.create_osint_workflow(target)
  
      logger.info(f"OSINT workflow created for {domain}")
  
      return workflow

• src/rest_api_server/managers.py:244-337 (helper)

  Core helper function in BugBountyWorkflowManager that generates the detailed OSINT workflow structure with multiple phases, tools, and estimated times.

  def create_osint_workflow(self, target: BugBountyTarget) -> dict[str, Any]:
      """Create OSINT (Open Source Intelligence) gathering workflow."""
      workflow = {
          "target": target.domain,
          "osint_phases": [],
          "estimated_time": 0,
          "data_sources": 0,
      }
  
      # Phase 1: Domain Intelligence
      domain_intel_phase = {
          "name": "domain_intelligence",
          "description": "Gather domain registration and ownership information",
          "tools": [
              {"tool": "whois", "params": {"domain": target.domain}},
              {"tool": "dnsenum", "params": {"domain": target.domain}},
              {"tool": "fierce", "params": {"domain": target.domain}},
          ],
          "expected_outputs": ["whois.txt", "dns_records.txt", "subdomains.txt"],
          "estimated_time": 120,
      }
      workflow["osint_phases"].append(domain_intel_phase)
  
      # Phase 2: Social Media & Web Presence
      social_phase = {
          "name": "social_intelligence",
          "description": "Gather social media and web presence intelligence",
          "tools": [
              {
                  "tool": "theharvester",
                  "params": {"domain": target.domain, "sources": "all"},
              },
              {
                  "tool": "search_engines",
                  "params": {"query": f"site:{target.domain}"},
              },
          ],
          "expected_outputs": [
              "emails.txt",
              "social_profiles.txt",
              "web_mentions.txt",
          ],
          "estimated_time": 180,
      }
      workflow["osint_phases"].append(social_phase)
  
      # Phase 3: Technology Stack Analysis
      tech_stack_phase = {
          "name": "technology_analysis",
          "description": "Identify technologies, frameworks, and infrastructure",
          "tools": [
              {"tool": "wappalyzer", "params": {"url": f"https://{target.domain}"}},
              {"tool": "builtwith", "params": {"domain": target.domain}},
              {"tool": "httpx", "params": {"tech_detect": True}},
          ],
          "expected_outputs": [
              "technologies.json",
              "frameworks.txt",
              "infrastructure.txt",
          ],
          "estimated_time": 90,
      }
      workflow["osint_phases"].append(tech_stack_phase)
  
      # Phase 4: Historical Data Analysis
      historical_phase = {
          "name": "historical_analysis",
          "description": "Analyze historical data and archived content",
          "tools": [
              {"tool": "waybackurls", "params": {"domain": target.domain}},
              {
                  "tool": "gau",
                  "params": {
                      "domain": target.domain,
                      "blacklist": (
                          "jpg,jpeg,gif,css,tif,tiff,png,ttf,woff,woff2,ico,pdf,svg,txt"
                      ),
                  },
              },
          ],
          "expected_outputs": ["historical_urls.txt", "archived_content.txt"],
          "estimated_time": 150,
      }
      workflow["osint_phases"].append(historical_phase)
  
      # Calculate totals
      workflow["estimated_time"] = sum(
          phase["estimated_time"] for phase in workflow["osint_phases"]
      )
      workflow["data_sources"] = sum(
          len(phase["tools"]) for phase in workflow["osint_phases"]
      )
  
      return workflow

• src/rest_api_server/managers.py:7-19 (schema)

  Dataclass defining the schema/structure for BugBountyTarget used in workflow generation, including domain and other parameters.

  @dataclass
  class BugBountyTarget:
      """Bug bounty target information."""
  
      domain: str
      scope: list[str] = field(default_factory=list)
      out_of_scope: list[str] = field(default_factory=list)
      program_type: str = "web"  # web, api, mobile, iot
      priority_vulns: list[str] = field(
          default_factory=lambda: ["rce", "sqli", "xss", "idor", "ssrf"]
      )
      bounty_range: str = "unknown"