Censorship infrastructure attribution: linking DPI deployments to procurement records, WHOIS data, and ASN registration history

from dataclasses import dataclass, field
from typing import Optional
import re

@dataclass
class DpiVendorSignature:
    vendor_name: str
    # RST timing range (None = RST not used by this vendor)
    rst_timing_min_ms: Optional[float]
    rst_timing_max_ms: Optional[float]
    # Known DNS injection IP prefixes (CIDR notation)
    injection_ip_prefixes: list[str]
    # Block page HTML SHA-256 hash prefixes (6 hex chars; partial match)
    block_page_hash_prefixes: list[str]
    # TLS CA fingerprints (SHA-256 of Subject Public Key Info)
    ca_spki_fingerprints: list[str]
    # Known deployment countries
    country_codes: list[str]
    # Procurement source URLs
    procurement_refs: list[str] = field(default_factory=list)
    # Confidence of attribution (0.0-1.0)
    signature_confidence: float = 0.85

# Example: Russia TSPU signature
TSPU_SIGNATURE = DpiVendorSignature(
    vendor_name='Ericsson TSPU (Russia)',
    rst_timing_min_ms=0.5,
    rst_timing_max_ms=3.0,
    injection_ip_prefixes=[],
    block_page_hash_prefixes=['a8f12c', 'd3e904'],
    ca_spki_fingerprints=[],   # RST-based; no TLS MITM
    country_codes=['RU'],
    procurement_refs=[
        'https://zakupki.gov.ru/epz/order/notice/ea20/view/common-info.html?regNumber=0373100049720000011',
        'https://zakupki.gov.ru/epz/order/notice/ea20/view/common-info.html?regNumber=0373100049721000008',
    ],
    signature_confidence=0.94,
)

def score_signature_match(
    measurement: dict,
    signature: DpiVendorSignature,
) -> float:
    """
    Compute a [0.0, 1.0] match score between a measurement and a DPI vendor signature.
    measurement keys: rst_timing_ms, block_page_hash, injection_ip, ca_spki
    """
    score = 0.0
    weight_sum = 0.0

    # RST timing match (weight: 0.35)
    if signature.rst_timing_min_ms is not None and 'rst_timing_ms' in measurement:
        weight_sum += 0.35
        t = measurement['rst_timing_ms']
        if signature.rst_timing_min_ms <= t <= signature.rst_timing_max_ms:
            score += 0.35
        elif t <= signature.rst_timing_max_ms * 1.5:
            score += 0.15   # partial credit for near-match

    # Block page hash match (weight: 0.30)
    if 'block_page_hash' in measurement and measurement['block_page_hash']:
        weight_sum += 0.30
        h = measurement['block_page_hash'][:6]
        if h in signature.block_page_hash_prefixes:
            score += 0.30

    # DNS injection IP match (weight: 0.25)
    if 'injection_ip' in measurement and measurement['injection_ip']:
        weight_sum += 0.25
        from ipaddress import ip_address, ip_network
        try:
            ip = ip_address(measurement['injection_ip'])
            for prefix in signature.injection_ip_prefixes:
                if ip in ip_network(prefix, strict=False):
                    score += 0.25
                    break
        except ValueError:
            pass

    # TLS CA SPKI match (weight: 0.10)
    if 'ca_spki' in measurement and measurement['ca_spki']:
        weight_sum += 0.10
        if measurement['ca_spki'] in signature.ca_spki_fingerprints:
            score += 0.10

    if weight_sum == 0:
        return 0.0

    return (score / weight_sum) * signature.signature_confidence

import httpx
import re
from datetime import datetime

PROCUREMENT_SOURCES = {
    'RU': {
        'base_url': 'https://zakupki.gov.ru/epz/order/extendedsearch/results.html',
        'keywords': ['ТСПУ', 'DPI', 'система управления трафиком', 'Ericsson'],
        'encoding': 'utf-8',
    },
    'TR': {
        'base_url': 'https://ekap.kik.gov.tr/EKAP/Ortak/IhaleArama/index.html',
        'keywords': ['içerik filtreleme', 'URL filtreleme', 'NetClean', 'BTK'],
        'encoding': 'utf-8',
    },
    'KZ': {
        'base_url': 'https://goszakup.gov.kz/ru/search/lots',
        'keywords': ['контентная фильтрация', 'DPI', 'система блокировки'],
        'encoding': 'utf-8',
    },
}

def extract_vendor_from_tender(text: str) -> list[str]:
    """Extract vendor name mentions from procurement tender PDF text."""
    VENDOR_PATTERNS = [
        r'Ericsson',
        r'Sandvine',
        r'NetClean',
        r'Cisco(?:s+IronPort|s+WSA)?',
        r'PacketLogic',
        r'Netsweeper',
        r'SmartFilter',
    ]
    found = []
    for pattern in VENDOR_PATTERNS:
        if re.search(pattern, text, re.IGNORECASE):
            found.append(re.search(pattern, text, re.IGNORECASE).group(0))
    return list(set(found))