A Shodan alternative

A Shodan alternative#

A Python implementation of an internet-wide port and banner search engine, the same kind as Shodan, Censys, and ZoomEye. The operator’s own scanner, indexer, and search API for use against ranges they own or have written authorization to scan.

Warning

Authorization required. Unauthorized internet-wide scans are illegal in most jurisdictions (CFAA, Computer Misuse Act, equivalents). This project is for the operator’s own ASN, authorized engagements, internal corporate space, or sanctioned research networks. See Disclaimers.

Architecture#

Three tiers, each independently scalable. Scanners discover open services; indexers store and index the results; the search API serves the operator’s queries.

        flowchart LR
    subgraph Scanners
        S1["Scanner worker<br/>(masscan / zmap)"]
        S2["Banner worker<br/>(scapy / httpx)"]
    end
    subgraph Storage
        Q[("Queue<br/>(Redis)")]
        DB[("Elasticsearch<br/>/ Postgres")]
    end
    subgraph API
        A["Search API<br/>(FastAPI)"]
    end
    T[("Targets<br/>(CIDR list)")] --> S1
    S1 --> Q
    Q --> S2
    S2 --> DB
    DB --> A
    Op([Operator]) -->|HTTP| A

Component

Job

Targets

The CIDR ranges or hostname list the operator authorises to scan.

Scanner workers

Stage-1 port discovery with masscan or zmap for speed; results are (ip, port) tuples.

Banner workers

Stage-2 connect, TLS handshake, HTTP request, protocol probe; produce service banners and metadata.

Queue

Redis or RabbitMQ that hands work from scanners to banner workers and tolerates back-pressure.

Index

Elasticsearch (default) or Postgres with JSONB. Indexes service banners by port, fingerprint, country, ASN.

Search API

FastAPI that translates operator queries into index searches, returns JSON.

Scanning pipeline#

The scan flow per IP.

        sequenceDiagram
    participant T as Targets CIDR
    participant M as masscan
    participant Q as Redis queue
    participant B as Banner worker
    participant E as Elasticsearch

    T->>M: list of CIDRs + ports
    M-->>Q: (ip, port) tuples (open)
    Q->>B: dequeue (ip, port)
    B->>B: connect, TLS handshake, protocol probe
    B->>B: fingerprint service (HTTP headers,<br/>SSH banner, TLS cert, snmp)
    B-->>E: index document {ip, port, service, banner,<br/>tls, geo, asn, fingerprinted_at}

Project layout#

shodan-alt/
├── pyproject.toml
├── README.md
├── docker-compose.yml      # elasticsearch + redis
└── src/
    ├── targets.py          # CIDR loader
    ├── scanner/            # stage-1
    │   ├── __init__.py
    │   └── masscan_run.py
    ├── banners/            # stage-2
    │   ├── __init__.py
    │   ├── worker.py
    │   ├── http.py
    │   ├── ssh.py
    │   ├── tls.py
    │   └── snmp.py
    ├── index/              # storage
    │   ├── __init__.py
    │   └── es_client.py
    └── api/                # frontend
        ├── __init__.py
        ├── main.py         # FastAPI app
        └── query.py        # query parser

Stage 1#

Port discovery with masscan (preferred; multi-million PPS) or zmap. Wrapped from Python so you can drive whole ranges programmatically.

# src/scanner/masscan_run.py
import asyncio, json
from pathlib import Path

async def run(cidr: str, ports: str, rate: int = 10_000,
              out: Path = Path("scan.json")):
    proc = await asyncio.create_subprocess_exec(
        "masscan", cidr,
        "-p", ports,
        "--rate", str(rate),
        "-oJ", str(out),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    await proc.wait()
    with out.open() as f:
        for line in f:
            line = line.strip().rstrip(",")
            if not line or line in ("[", "]"):
                continue
            yield json.loads(line)

A producer that pushes (ip, port) onto the queue.

import redis.asyncio as redis

async def main():
    r = redis.Redis()
    async for hit in run("198.51.100.0/24", "1-1000"):
        for p in hit["ports"]:
            await r.lpush("banner_queue",
                          f"{hit['ip']}:{p['port']}")

Stage 2#

Banner workers pull from the queue, connect to each (ip, port), probe the protocol, and emit a banner record.

# src/banners/worker.py
import asyncio, json, time
from datetime import datetime
import redis.asyncio as redis
from elasticsearch import AsyncElasticsearch
from .http import probe_http
from .ssh  import probe_ssh
from .tls  import probe_tls

PROBES = {
    21: ("ftp",  lambda h, p: probe_tls(h, p)),
    22: ("ssh",  probe_ssh),
    80: ("http", lambda h, p: probe_http(h, p, "http")),
    443: ("https", lambda h, p: probe_http(h, p, "https")),
}

async def worker():
    r = redis.Redis()
    es = AsyncElasticsearch(["http://localhost:9200"])
    while True:
        job = await r.brpop("banner_queue", timeout=5)
        if not job:
            continue
        ip, port = job[1].decode().split(":")
        port = int(port)
        service, probe = PROBES.get(port, ("unknown", probe_tls))
        try:
            banner = await asyncio.wait_for(probe(ip, port), timeout=5)
        except Exception:
            continue
        doc = {
            "ip": ip, "port": port,
            "service": service, "banner": banner,
            "fingerprinted_at": datetime.utcnow().isoformat(),
        }
        await es.index(index="services", document=doc)

A protocol probe.

# src/banners/http.py
import httpx

async def probe_http(host: str, port: int, scheme: str) -> dict:
    url = f"{scheme}://{host}:{port}/"
    async with httpx.AsyncClient(verify=False, timeout=5) as c:
        r = await c.get(url, follow_redirects=False)
        return {
            "status":  r.status_code,
            "server":  r.headers.get("server"),
            "title":   _title(r.text),
            "headers": dict(r.headers),
        }

def _title(html: str) -> str | None:
    import re
    m = re.search(r"<title[^>]*>([^<]+)</title>",
                  html, re.IGNORECASE)
    return m.group(1).strip() if m else None

Search API#

FastAPI fronts Elasticsearch.

# src/api/main.py
from fastapi import FastAPI, Query
from elasticsearch import AsyncElasticsearch

app = FastAPI()
es  = AsyncElasticsearch(["http://localhost:9200"])

@app.get("/search")
async def search(q: str = Query(...), size: int = 50):
    """Query DSL: 'port:443 country:DE server:nginx'."""
    must = []
    for part in q.split():
        if ":" in part:
            field, value = part.split(":", 1)
            must.append({"match": {field: value}})
    body = {"query": {"bool": {"must": must or [{"match_all": {}}]}}}
    res = await es.search(index="services", body=body, size=size)
    return [h["_source"] for h in res["hits"]["hits"]]

Common Tasks#

Stand up the infrastructure.

$ docker compose up -d elasticsearch redis

Run a stage-1 scan over an authorized CIDR.

$ sudo masscan 198.51.100.0/24 -p 22,80,443 \
    --rate 50000 -oJ scan.json
$ uv run python -m scanner.masscan_run \
    --cidr 198.51.100.0/24 --ports 1-1000

Start banner workers.

$ uv run python -m banners.worker &
$ uv run python -m banners.worker &
$ uv run python -m banners.worker &

Serve the search API.

$ uv run uvicorn api.main:app --host 0.0.0.0 --port 8000

Query the index.

$ curl 'http://localhost:8000/search?q=port:443+server:nginx'

Re-index when the operator changes the mapping.

$ uv run python -m index.reindex \
    --src services --dst services-v2

Backfill geolocation from MaxMind.

$ uv run python -m index.enrich --geoip GeoLite2-City.mmdb

References#

Contents