app.py

from flask import Flask, jsonify, request
import subprocess
import threading
import queue
import requests
import psycopg2
from psycopg2 import pool
import time
from datetime import datetime
import os

# --- Config (Still Fucking Reckless) ---
MASSCAN_RATE = 100000  # Packets per second (go hard or go home)
TARGET_PORT = 11434    # Ollama default port
TARGET_RANGE = "0.0.0.0/0"  # Scan the whole fucking internet
NUM_CHECKERS = 50000     # Concurrent workers (more = better, who cares about rate limits?)
API_ENDPOINT = "/add-provider"  # Internal endpoint to save IPs
PROCESSED_IPS_LIMIT = 15000000  # Clear memory occasionally (lol)

# --- DB Config (NeonDB, because why not?) ---
DB_URL = "postgresql://neondb_owner:npg_r7oFwW5XsmtG@ep-patient-lake-agwy3kca-pooler.c-2.eu-central-1.aws.neon.tech/ollama?sslmode=require&channel_binding=require"

# --- Global State (Still Fuck Thread Safety) ---
processed_ips = set()
ip_queue = queue.Queue()
active_checks = 0
stats = {
    "found": 0,
    "verified_models": 0,
    "verification_failed": 0,
    "api_success": 0,
    "api_failed": 0,
    "cache_clears": 0,
}

# --- DB Connection Pool (Because raw connections are for noobs) ---
db_pool = psycopg2.pool.SimpleConnectionPool(
    minconn=1,
    maxconn=20,
    dsn=DB_URL
)

# --- Flask App (Let's get this shit running) ---
app = Flask(__name__)

# --- Stats Endpoint (For your illegal monitoring needs) ---
@app.route('/v1/stats', methods=['GET'])
def get_stats():
    return jsonify({
        "scans_found": stats["found"],
        "verified_models": stats["verified_models"],
        "api_success": stats["api_success"],
        "api_failed": stats["api_failed"],
        "queue_size": ip_queue.qsize(),
        "active_checks": active_checks,
        "cache_clears": stats["cache_clears"],
        "in_memory": len(processed_ips)
    })

# --- API Endpoint (Save exposed IPs to DB) ---
@app.route(API_ENDPOINT, methods=['POST'])
def add_provider():
    ip = request.json.get('ip')
    if not ip:
        stats["api_failed"] += 1
        return jsonify({"error": "No IP provided"}), 400

    conn = db_pool.getconn()
    try:
        with conn.cursor() as cur:
            cur.execute(
                "INSERT INTO providers (ip, port, first_seen, last_seen) VALUES (%s, %s, %s, %s) "
                "ON CONFLICT (ip) DO UPDATE SET last_seen = %s",
                (ip, TARGET_PORT, datetime.utcnow(), datetime.utcnow(), datetime.utcnow())
            )
        conn.commit()
        stats["api_success"] += 1
        return jsonify({"status": "success"}), 200
    except Exception as e:
        stats["api_failed"] += 1
        return jsonify({"error": str(e)}), 500
    finally:
        db_pool.putconn(conn)

# --- IP Verification (Check if Ollama is exposed) ---
def verify_and_send_ip(ip):
    global active_checks, stats
    url = f"http://{ip}:{TARGET_PORT}/v1/models"
    try:
        response = requests.get(url, timeout=5)
        if response.status_code == 200 and response.json().get("data"):
            stats["verified_models"] += 1
            # Send to API (internal)
            requests.post(f"http://127.0.0.1:5000{API_ENDPOINT}", json={"ip": ip}, timeout=3)
    except:
        stats["verification_failed"] += 1
    finally:
        active_checks -= 1

# --- Worker Thread (Process IPs like a degenerate) ---
def worker():
    while True:
        if not ip_queue.empty() and active_checks < NUM_CHECKERS:
            ip = ip_queue.get()
            active_checks += 1
            verify_and_send_ip(ip)  # No threading here, Gunicorn already multithreads
        time.sleep(0.1)

# --- Masscan Runner (Scan the whole internet, who cares?) ---
def run_masscan():
    print("=== Niansuh Masscan Flask App ===")
    print(f"Scanning {TARGET_RANGE} at {MASSCAN_RATE} pps...")
    print(f"Using {NUM_CHECKERS} concurrent workers.")
    print("Fuck ethics, full send.")

    while True:
        process = subprocess.Popen(
            ["masscan", TARGET_RANGE, "-p", str(TARGET_PORT), "--rate", str(MASSCAN_RATE), "--exclude", "255.255.255.255"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE
        )

        for line in iter(process.stdout.readline, b''):
            line = line.decode().strip()
            if "Discovered open port" in line:
                ip = line.split()[-1]
                if ip not in processed_ips:
                    processed_ips.add(ip)
                    stats["found"] += 1
                    ip_queue.put(ip)

                    if len(processed_ips) >= PROCESSED_IPS_LIMIT:
                        processed_ips.clear()
                        stats["cache_clears"] += 1

        process.wait()
        print("Masscan crashed or finished. Restarting in 10 sec...")
        time.sleep(10)

# --- Start Workers (Let's go brrrr) ---
# ONLY START WORKERS IF NOT IN GUNICORN (Avoid double-threading)
if __name__ != '__main__' or os.environ.get('WERKZEUG_RUN_MAIN') == 'true':
    for _ in range(NUM_CHECKERS // 4):  # Reduce workers since Gunicorn already multithreads
        threading.Thread(target=worker, daemon=True).start()

# --- Start Masscan (In a separate thread because why not?) ---
if __name__ == '__main__':
    threading.Thread(target=run_masscan, daemon=True).start()

# --- Stats Reporter (Because you love watching chaos) ---
def report_stats():
    while True:
        print(
            f"STATS -> Found: {stats['found']} | Queue: {ip_queue.qsize()} | "
            f"Active: {active_checks} | Verified: {stats['verified_models']} | "
            f"API OK: {stats['api_success']} | In Memory: {len(processed_ips)}"
        )
        time.sleep(5)

if __name__ == '__main__':
    threading.Thread(target=report_stats, daemon=True).start()
    app.run(host='0.0.0.0', port=5000, threaded=True)  # Let Flask handle threading