Tor Exit Node Detection Database

prepare(
        "SELECT node_type, nickname, country, is_active
         FROM tor_nodes
         WHERE ip_address = :ip
           AND node_type = 'exit_node'"
    );
    $stmt->execute(['ip' => $ip]);
    return $stmt->fetch(PDO::FETCH_ASSOC) ?: null;
}

$ip = $_SERVER['REMOTE_ADDR'];
$tor = is_tor_exit($pdo, $ip);

if ($tor) {
    // Traffic is coming from a Tor exit node
    if ($tor['is_active']) {
        // Active exit node - apply your Tor policy
        // Options: block, require CAPTCHA, require 2FA, or just log
        log_security_event('tor_access', [
            'ip' => $ip,
            'node' => $tor['nickname'],
            'country' => $tor['country'],
        ]);

        // Example: require additional verification for sensitive actions
        if (is_sensitive_action()) {
            require_two_factor_auth();
        }
    }
}

import redis
import csv

# One-time: load the AntiProxies Tor CSV into Redis
# Using a Redis hash gives O(1) lookups per IP

def load_tor_nodes(redis_client, csv_path: str):
    """Load Tor exit nodes into Redis for fast lookups."""
    with open(csv_path) as f:
        reader = csv.DictReader(f)
        pipe = redis_client.pipeline()
        count = 0
        for row in reader:
            if row["node_type"] == "exit_node":
                pipe.hset(f"tor:{row['ip_address']}", mapping={
                    "node_type": row["node_type"],
                    "country": row["country"],
                    "is_active": row["is_active"],
                    "nickname": row["nickname"],
                })
                count += 1
        pipe.execute()
    print(f"Loaded {count} Tor exit nodes into Redis")

def is_tor_exit(redis_client, ip: str) -> dict | None:
    """Check if an IP is a known Tor exit node. O(1) lookup."""
    data = redis_client.hgetall(f"tor:{ip}")
    return {k.decode(): v.decode() for k, v in data.items()} if data else None

# Usage in your web framework
r = redis.Redis()
tor_info = is_tor_exit(r, request.remote_addr)
if tor_info and tor_info["is_active"] == "1":
    # Apply Tor-specific security policy
    require_extra_verification()