diff --git a/.gitignore b/.gitignore index 62c9581..15223c3 100644 --- a/.gitignore +++ b/.gitignore @@ -65,6 +65,7 @@ local_packages/*/.dart_tool/ # Secrets android/key.properties android/admob.properties +AGENTS.md SERVER_NOTES.md lib/logic/backup/ 00-FULL-REPORT.md diff --git a/lib/logic/ad_manager.dart b/lib/logic/ad_manager.dart index 5efdbb0..62373e3 100644 --- a/lib/logic/ad_manager.dart +++ b/lib/logic/ad_manager.dart @@ -1,5 +1,6 @@ import 'dart:async'; import 'dart:convert'; +import 'dart:math'; import 'package:flutter/foundation.dart'; import 'package:google_mobile_ads/google_mobile_ads.dart'; import 'package:paladinvpn/logic/crypto_service.dart'; @@ -102,12 +103,19 @@ class AdManager extends ChangeNotifier { final keys = await CryptoService.getOrCreateKeys(); final pubKey = keys['publicKey']!; - // 3. Build SSV options to send to our Hivemind server via Google's callback + // 3. Generate a connect nonce so the subsequent poll can verify + // it's talking to the session created by *this* ad. + final nonce = '${Random().nextInt(0x7FFFFFFF)}-${DateTime.now().millisecondsSinceEpoch}'; + HivemindService.setExpectedNonce(nonce); + debugPrint('[AdManager] Ad nonce: $nonce'); + + // 4. Build SSV options to send to our Hivemind server via Google's callback final ssvOptions = ServerSideVerificationOptions( customData: jsonEncode({ 'device_id': deviceId, 'public_key': pubKey, 'ad_type': adType, + 'nonce': nonce, }), ); diff --git a/lib/logic/hivemind_service.dart b/lib/logic/hivemind_service.dart index 08a3ae7..ec3b2aa 100644 --- a/lib/logic/hivemind_service.dart +++ b/lib/logic/hivemind_service.dart @@ -1,22 +1,46 @@ import 'dart:async'; import 'dart:convert'; +import 'dart:math'; import 'package:flutter/foundation.dart'; import 'package:http/http.dart' as http; import 'package:paladinvpn/logic/app_config.dart'; import 'package:paladinvpn/logic/crypto_service.dart'; class HivemindService { + /// Nonce generated at the start of a connect cycle. + /// Sent to the server during session creation and verified in every poll + /// response so we never accidentally resume a stale session. + static String? _expectedNonce; + + /// Stores a nonce so that fetchConfigWithPolling can verify it. + /// Called by AdManager.showAd() before showing a rewarded ad so the + /// subsequent poll accepts only the session created by *this* ad. + static void setExpectedNonce(String nonce) { + _expectedNonce = nonce; + } + /// Polls the Hivemind server until the session is ready or timeout occurs. /// Returns the WireGuard config string. /// /// [onAttempt] is called after each poll with (current, total) so the UI /// can show progress (e.g. "Contacting server (3/15)…"). + static Future fetchConfigWithPolling({ void Function(int attempt, int total)? onAttempt, }) async { final deviceId = await CryptoService.getDeviceId(); final keys = await CryptoService.getOrCreateKeys(); final privKey = keys['privateKey']!; + + // ── Generate a connect nonce ────────────────────────────────────── + // Every fresh connect cycle gets a new random token. The server + // stores it in the session and echoes it back in /session/status. + // If the poll returns a session whose nonce differs (e.g. a stale + // leftover from a failed disconnectAndCleanup), we skip it and keep + // polling until the right session appears or the timeout fires. + final nonce = '${Random().nextInt(0x7FFFFFFF)}-${DateTime.now().millisecondsSinceEpoch}'; + _expectedNonce = nonce; + debugPrint('[HivemindService] Connect nonce: $nonce'); final url = Uri.parse('${AppConfig.hivemindApiBase}/session/status?device_id=$deviceId'); @@ -30,6 +54,7 @@ class HivemindService { 'device_id': deviceId, 'public_key': keys['publicKey']!, 'ad_type': 'main_ad', + 'nonce': nonce, }); final fakeAdmobPingUrl = Uri.parse( '${AppConfig.hivemindApiBase}/admob/callback?signature=test&key_id=test&custom_data=${Uri.encodeComponent(customData)}'); @@ -47,9 +72,17 @@ class HivemindService { if (response.statusCode == 200) { final data = jsonDecode(response.body); - debugPrint('[HivemindService] Session active, constructing config…'); - - if (data['active'] == true && data['client_ip'] != null && data['server_pubkey'] != null) { + debugPrint('[HivemindService] Session found, checking nonce…'); + + // ── Nonce gate ────────────────────────────────────────────── + // Reject any session whose nonce doesn't match what we sent. + // If the server is old and doesn't return a nonce we still + // accept it (backwards-compat), but a mismatch is a hard skip. + final serverNonce = data['nonce'] as String?; + if (_expectedNonce != null && serverNonce != null && serverNonce != _expectedNonce) { + debugPrint('[HivemindService] Nonce mismatch (expected $_expectedNonce, got $serverNonce) — stale session, retrying…'); + // Fall through to retry — do NOT accept this session. + } else if (data['active'] == true && data['client_ip'] != null && data['server_pubkey'] != null) { final clientIp = data['client_ip']; final clientIpv6 = data['client_ipv6'] ?? ''; final serverPubkey = data['server_pubkey']; @@ -82,6 +115,7 @@ AllowedIPs = ${AppConfig.allowedIps} Endpoint = ${AppConfig.serverEndpoint} PersistentKeepalive = ${AppConfig.persistentKeepalive} '''; + _expectedNonce = null; // consumed successfully return configText; } } else { diff --git a/server_hivemind_5_0.py b/server_hivemind_5_0.py index 4ccb3d0..fb2eb94 100644 --- a/server_hivemind_5_0.py +++ b/server_hivemind_5_0.py @@ -3,6 +3,18 @@ Hivemind VPN Session Manager Manages time + data quotas for AmneziaWG peers via ad rewards. Designed for Android-only clients. + +=== TABLE OF CONTENTS =========================================================== + LINE SECTION + 44 Configuration — ad rewards, throttle rate, check interval + 60 State & Cache — sessions dict, IP pool, server pubkey + 93 Flask Routes — /admob/callback, /session/status, /session/stop + 220 Session Logic — _start_or_extend_session (create / extend) + 295 AdMob SSV Verification — Google ECDSA signature checks + 380 Management Loop — expiry + data-quota watchdog (60 s tick) + 426 Low-Level Plumbing — awg commands, tc throttling, IP pool, peers + 567 Entry Point — daemon thread + gunicorn bootstrap +================================================================================ """ import time @@ -13,11 +25,12 @@ import requests import shlex import re import urllib.parse +import ipaddress +import base64 from flask import Flask, request, jsonify from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.serialization import load_der_public_key, load_pem_public_key -import base64 app = Flask(__name__) @@ -44,6 +57,15 @@ BONUS_AD_BYTES = 1 * 1024 * 1024 * 1024 # 1 GB THROTTLE_RATE = "1.5mbit" +# AdMob SSV verification toggle +# True = dev mode, accept any callback (no signature check) +# False = prod mode, enforce Google ECDSA signature verification +# +# ⚠️ Set to False AFTER Google Play verifies your AdMob implementation is +# working correctly with real rewarded ads. Until then, keep True so +# the debug fake-callback path in the Flutter app still functions. +ADMOB_BYPASS = True + # ============================================================================== # STATE & CACHE # ============================================================================== @@ -61,7 +83,8 @@ SERVER_PUBKEY = None # "expires_at": 1716480000, # "quota_bytes": 2147483648, # "baseline_bytes": 0, -# "is_throttled": False +# "is_throttled": False, +# "nonce": "a7f3b2-1716480000000", # } # } active_sessions = {} @@ -70,10 +93,342 @@ active_sessions = {} IP_POOL_START = 100 IP_POOL_END = 250 +# Pre-computed pool of all available IPv4 addresses in the 10.8.0.0/16 subnet. +# Initialised once at module load; allocate_ip() pops from it, release_ip() returns to it. +AVAILABLE_IPS = {str(ip) for ip in ipaddress.IPv4Network('10.8.0.0/16').hosts()} +AVAILABLE_IPS.discard('10.8.0.1') + # ============================================================================== -# SYSTEM COMMAND HELPERS +# FLASK ROUTES (API surface — what the outside world sees) # ============================================================================== + +@app.route('/api/admob/callback', methods=['GET']) +def admob_callback(): + """ + Google AdMob SSV callback — called server-to-server when a rewarded ad completes. + The Android client never touches this endpoint; Google calls it directly. + + custom_data (JSON, set by the app before showing the ad): + { "device_id": "...", "public_key": "...", "ad_type": "main_ad"|"bonus_ad", "nonce": "..." } + """ + try: + if ADMOB_BYPASS: + # ── DEV: accept any callback without signature verification ── + custom_data_str = request.args.get('custom_data') + if not custom_data_str: + return "OK", 200 + else: + # ── PROD: enforce Google ECDSA signature verification ──────── + signature = request.args.get('signature') + key_id = request.args.get('key_id') + custom_data_str = request.args.get('custom_data') + + # Google sends a verification ping with signature+key_id but no custom_data + if not custom_data_str: + return "OK", 200 + + if not all([signature, key_id]): + return "Missing parameters", 400 + + raw_qs = request.query_string.decode("utf-8") + if not _verify_admob_signature(raw_qs, signature, key_id): + print("[ADMOB] Rejected callback — invalid signature.") + return "Forbidden", 403 + + custom_data = json.loads(custom_data_str) + device_id = custom_data.get('device_id') + client_pubkey = custom_data.get('public_key') + + if not device_id or not client_pubkey: + return "Invalid custom_data payload", 400 + + ad_type = custom_data.get('ad_type', 'main_ad') + nonce = custom_data.get('nonce', None) + + success, msg = _start_or_extend_session(device_id, client_pubkey, ad_type, nonce) + return ("OK", 200) if success else (msg, 500) + + except Exception as e: + print(f"[ADMOB] SSV Error: {e}") + return "Internal Error", 500 + + +@app.route('/api/session/status', methods=['GET']) +def session_status(): + """ + Query: ?device_id=esf-xxx + Returns current session state so the Android app can construct its AmneziaWG config. + """ + device_id = request.args.get('device_id') + if not device_id: + return jsonify({"error": "Missing query parameter: device_id"}), 400 + + with sessions_lock: + session = active_sessions.get(device_id) + + if not session: + return jsonify({"active": False, "message": "No session found."}), 404 + + # Bulletproof fallback: If server public key is missing, try fetching it right now + global SERVER_PUBKEY + if not SERVER_PUBKEY: + pubkey = run_cmd("awg show awg0 public-key") + if pubkey: + SERVER_PUBKEY = pubkey + print(f"[AWG] Dynamic Server Public Key cached from awg0: {SERVER_PUBKEY}") + else: + SERVER_PUBKEY = SERVER_PUBKEY_FALLBACK or "server key=" + + stats = get_wireguard_stats() + current_bytes = stats.get(session["ip"], {}).get("total_bytes", 0) + used_bytes = max(0, current_bytes - session["baseline_bytes"]) + remaining_data = max(0, session["quota_bytes"] - used_bytes) + remaining_secs = max(0, session["expires_at"] - time.time()) + + return jsonify({ + "active": remaining_secs > 0, + "client_ip": session["ip"], + "client_ipv6": session.get("ipv6", ""), + "server_pubkey": SERVER_PUBKEY, + "is_throttled": session["is_throttled"], + "expires_in_seconds": int(remaining_secs), + "quota_bytes": session["quota_bytes"], + "used_bytes": used_bytes, + "remaining_bytes": remaining_data, + "nonce": session.get("nonce", ""), + }) + + +@app.route('/api/session/stop', methods=['POST']) +def session_stop(): + """Allows the app to explicitly tear down its session on disconnect.""" + data = request.get_json(silent=True) + device_id = data.get('device_id') if data else None + + if device_id: + with sessions_lock: + session = active_sessions.pop(device_id, None) + if session: + remove_wg_peer(session["pubkey"]) + if session["is_throttled"]: + remove_throttle(session["ip"]) + release_ip(session["ip"]) + print(f"[API] Explicit stop for {device_id}.") + + return jsonify({"success": True}) + + +# ============================================================================== +# SESSION LOGIC +# ============================================================================== + +def _start_or_extend_session(device_id, client_pubkey, ad_type, nonce=None): + """Core logic to provision or extend a session. + + nonce -- client-generated random token; stored in session and echoed + back in /session/status so the client can reject stale sessions. + """ + if not re.match(r"^[A-Za-z0-9+/=]{43,44}$", client_pubkey): + return False, "Invalid WireGuard public key format" + + now = time.time() + + with sessions_lock: + if device_id in active_sessions and ad_type == 'bonus_ad': + # Extend existing session + session = active_sessions[device_id] + session["expires_at"] += (BONUS_AD_MINUTES * 60) + session["quota_bytes"] += BONUS_AD_BYTES + + if session["is_throttled"]: + remove_throttle(session["ip"]) + session["is_throttled"] = False + + stats = get_wireguard_stats() + current_bytes = stats.get(session["ip"], {}).get("total_bytes", 0) + session["baseline_bytes"] = current_bytes + + print(f"[SESSION] Bonus time added for {device_id} (IP: {session['ip']}).") + return True, "Bonus time and data added." + + else: + # Create a brand new session (Main Ad) + if not validate_wg_pubkey(client_pubkey): + print(f"[SESSION] Rejected invalid pubkey from {device_id[:8]}...") + return False, "Invalid public key format." + + existing = active_sessions.get(device_id) + if existing: + if existing["is_throttled"]: + remove_throttle(existing["ip"]) + remove_wg_peer(existing["pubkey"]) + release_ip(existing["ip"]) + + ipv4 = allocate_ip() + if not ipv4: + return False, "Server full! No available IP addresses." + + ipv6 = ipv4_to_ipv6(ipv4) + add_wg_peer(client_pubkey, ipv4, ipv6) + + # Fetch immediate stats to set baseline + # (Though brand new peers usually start at 0) + stats = get_wireguard_stats() + current_bytes = stats.get(ipv4, {}).get("total_bytes", 0) + + active_sessions[device_id] = { + "ip": ipv4, + "ipv6": ipv6, + "pubkey": client_pubkey, + "expires_at": now + (MAIN_AD_MINUTES * 60), + "quota_bytes": MAIN_AD_BYTES, + "baseline_bytes": current_bytes, + "is_throttled": False, + "nonce": nonce, + } + print(f"[SESSION] Main session started for {device_id} (IP: {ipv4} / {ipv6}).") + return True, "Main session active." + + + + + +# ============================================================================== +# ADMOB SERVER-TO-SERVER (SSV) VERIFICATION +# ============================================================================== +ADMOB_KEYS_URL = "https://www.gstatic.com/admob/reward/verifier-keys.json" + +# Cache: { key_id_str: EllipticCurvePublicKey } +# Refreshed at most once every 24 hours so we don't hammer Google's CDN. +_admob_key_cache = {} +_admob_keys_fetched_at = 0.0 +_ADMOB_KEY_TTL = 86400 # seconds + + +def _get_admob_public_key(key_id: str): + """ + Returns the Google ECDSA public-key object for the given key_id. + Fetches and caches the key list from gstatic; refreshes every 24h. + Returns None if the key_id is unknown. + """ + global _admob_key_cache, _admob_keys_fetched_at + + now = time.time() + if now - _admob_keys_fetched_at > _ADMOB_KEY_TTL or not _admob_key_cache: + try: + resp = requests.get(ADMOB_KEYS_URL, timeout=5) + resp.raise_for_status() + data = resp.json() + + new_cache = {} + for entry in data.get("keys", []): + kid = str(entry["keyId"]) + pem = entry["pem"].encode() + new_cache[kid] = load_pem_public_key(pem) + + _admob_key_cache = new_cache + _admob_keys_fetched_at = now + print(f"[ADMOB] Refreshed {len(new_cache)} public key(s) from Google.") + + except Exception as e: + print(f"[ADMOB] WARNING: Could not refresh public keys: {e}") + # Fall through with stale cache rather than rejecting all valid callbacks + + return _admob_key_cache.get(key_id) + + +def _verify_admob_signature(raw_query_string: str, signature_b64url: str, key_id: str) -> bool: + """ + Verifies an AdMob SSV ECDSA-SHA256 signature. + Google signs all query params EXCEPT 'signature' and 'key_id', in original order. + """ + try: + public_key = _get_admob_public_key(key_id) + if not public_key: + print("[ADMOB] Unknown key_id — rejecting callback.") + return False + + # Reconstruct the signed payload by stripping signature + key_id from raw qs + parsed = urllib.parse.parse_qs(raw_query_string, keep_blank_values=True) + + # Google signs query params in the exact order they appear in the URL. + # Use the raw string to preserve ordering — parse_qs loses it. + ordered_pairs = [] + for pair in raw_query_string.split('&'): + if '=' not in pair: + continue + k, v = pair.split('=', 1) + if k in ('signature', 'key_id'): + continue + ordered_pairs.append(f"{k}={v}") + payload = '&'.join(ordered_pairs).encode('utf-8') + + # Google uses URL-safe base64 (no padding) for the signature + sig_bytes = base64.urlsafe_b64decode(signature_b64url + '==') + + public_key.verify( + sig_bytes, + payload, + ec.ECDSA(hashes.SHA256()) + ) + return True + + except Exception as e: + print(f"[ADMOB] Signature verification failed: {e}") + return False + + +# ============================================================================== +# MANAGEMENT LOOP (The Hivemind — expiry + data-quota watchdog) +# ============================================================================== +def management_loop(): + print(f"[HIVEMIND] Started. Tick every {CHECK_INTERVAL_SECONDS}s.") + init_wireguard() + + while True: + try: + stats = get_wireguard_stats() + now = time.time() + + with sessions_lock: + expired_devices = [] + + for device_id, session in list(active_sessions.items()): + ip = session["ip"] + + # ── 1. Check expiry ─────────────────────────────────────── + if now > session["expires_at"]: + print(f"[HIVEMIND] Session expired for {device_id} ({ip}). Removing peer.") + if session["is_throttled"]: + remove_throttle(ip) + remove_wg_peer(session["pubkey"]) + release_ip(ip) + expired_devices.append(device_id) + continue + + # ── 2. Check data quota ─────────────────────────────────── + if ip in stats: + used_bytes = stats[ip]["total_bytes"] - session["baseline_bytes"] + + if used_bytes > session["quota_bytes"] and not session["is_throttled"]: + print(f"[HIVEMIND] Data cap reached for {ip} ({used_bytes/1e9:.2f} GB used). Throttling.") + apply_throttle(ip) + session["is_throttled"] = True + + for device_id in expired_devices: + del active_sessions[device_id] + + except Exception as e: + print(f"[HIVEMIND] Unhandled error: {e}") + + time.sleep(CHECK_INTERVAL_SECONDS) + + +# ============================================================================== +# LOW-LEVEL PLUMBING (awg, tc, IP pool, peer management) +# ============================================================================== + def run_cmd(cmd, ignore_errors=False): """Executes a shell command directly on the host (no Docker middleman).""" # All commands run directly on the host against awg0. @@ -177,25 +532,12 @@ def get_wireguard_stats(): return stats -import ipaddress - -# ============================================================================== -# IPv6 HELPERS -# ============================================================================== def ipv4_to_ipv6(ipv4): """Map 10.8.X.Y → fd42::X:Y (ULA for internal VPN routing, no NAT64 needed).""" parts = ipv4.split('.') return f"fd42::{parts[2]}:{parts[3]}" -# ============================================================================== -# IP POOL MANAGEMENT -# ============================================================================== -# The server subnet is 10.8.0.0/16, giving ~65,530 IPs for public use. -#60 THOUSAND OR 64 THOUSAND ARGUEMENT GOES HERE -AVAILABLE_IPS = {str(ip) for ip in ipaddress.IPv4Network('10.8.0.0/16').hosts()} -AVAILABLE_IPS.discard('10.8.0.1') -# We need a fallback if we somehow exhaust them, though unlikely def allocate_ip(): """Pops an available IP from the pre-generated pool.""" if not AVAILABLE_IPS: @@ -224,320 +566,6 @@ def remove_wg_peer(pubkey): print(f"[AWG] Ephemeral peer removed: {pubkey[:8]}...") -# ============================================================================== -# BACKGROUND MANAGEMENT LOOP (The Hivemind) -# ============================================================================== -def management_loop(): - print(f"[HIVEMIND] Started. Tick every {CHECK_INTERVAL_SECONDS}s.") - init_wireguard() - - while True: - try: - stats = get_wireguard_stats() - now = time.time() - - with sessions_lock: - expired_devices = [] - - for device_id, session in list(active_sessions.items()): - ip = session["ip"] - - # ── 1. Check expiry ─────────────────────────────────────── - if now > session["expires_at"]: - print(f"[HIVEMIND] Session expired for {device_id} ({ip}). Removing peer.") - if session["is_throttled"]: - remove_throttle(ip) - remove_wg_peer(session["pubkey"]) - release_ip(ip) - expired_devices.append(device_id) - continue - - # ── 2. Check data quota ─────────────────────────────────── - if ip in stats: - used_bytes = stats[ip]["total_bytes"] - session["baseline_bytes"] - - if used_bytes > session["quota_bytes"] and not session["is_throttled"]: - print(f"[HIVEMIND] Data cap reached for {ip} ({used_bytes/1e9:.2f} GB used). Throttling.") - apply_throttle(ip) - session["is_throttled"] = True - - for device_id in expired_devices: - del active_sessions[device_id] - - except Exception as e: - print(f"[HIVEMIND] Unhandled error: {e}") - - time.sleep(CHECK_INTERVAL_SECONDS) - - -# ============================================================================== -# SESSION LOGIC -# ============================================================================== -def _start_or_extend_session(device_id, client_pubkey, ad_type): - """Core logic to provision or extend a session.""" - if not re.match(r"^[A-Za-z0-9+/=]{43,44}$", client_pubkey): - return False, "Invalid WireGuard public key format" - - now = time.time() - - with sessions_lock: - if device_id in active_sessions and ad_type == 'bonus_ad': - # Extend existing session - session = active_sessions[device_id] - session["expires_at"] += (BONUS_AD_MINUTES * 60) - session["quota_bytes"] += BONUS_AD_BYTES - - if session["is_throttled"]: - remove_throttle(session["ip"]) - session["is_throttled"] = False - - stats = get_wireguard_stats() - current_bytes = stats.get(session["ip"], {}).get("total_bytes", 0) - session["baseline_bytes"] = current_bytes - - print(f"[SESSION] Bonus time added for {device_id} (IP: {session['ip']}).") - return True, "Bonus time and data added." - - else: - # Create a brand new session (Main Ad) - if not validate_wg_pubkey(client_pubkey): - print(f"[SESSION] Rejected invalid pubkey from {device_id[:8]}...") - return False, "Invalid public key format." - - existing = active_sessions.get(device_id) - if existing: - if existing["is_throttled"]: - remove_throttle(existing["ip"]) - remove_wg_peer(existing["pubkey"]) - release_ip(existing["ip"]) - - ipv4 = allocate_ip() - if not ipv4: - return False, "Server full! No available IP addresses." - - ipv6 = ipv4_to_ipv6(ipv4) - add_wg_peer(client_pubkey, ipv4, ipv6) - - # Fetch immediate stats to set baseline - # (Though brand new peers usually start at 0) - stats = get_wireguard_stats() - current_bytes = stats.get(ipv4, {}).get("total_bytes", 0) - - active_sessions[device_id] = { - "ip": ipv4, - "ipv6": ipv6, - "pubkey": client_pubkey, - "expires_at": now + (MAIN_AD_MINUTES * 60), - "quota_bytes": MAIN_AD_BYTES, - "baseline_bytes": current_bytes, - "is_throttled": False, - } - print(f"[SESSION] Main session started for {device_id} (IP: {ipv4} / {ipv6}).") - return True, "Main session active." - - - - - -# ============================================================================== -# ADMOB SERVER-TO-SERVER (SSV) VERIFICATION -# ============================================================================== -ADMOB_KEYS_URL = "https://www.gstatic.com/admob/reward/verifier-keys.json" - -# Cache: { key_id_str: EllipticCurvePublicKey } -# Refreshed at most once every 24 hours so we don't hammer Google's CDN. -_admob_key_cache = {} -_admob_keys_fetched_at = 0.0 -_ADMOB_KEY_TTL = 86400 # seconds - - -def _get_admob_public_key(key_id: str): - """ - Returns the Google ECDSA public-key object for the given key_id. - Fetches and caches the key list from gstatic; refreshes every 24h. - Returns None if the key_id is unknown. - """ - global _admob_key_cache, _admob_keys_fetched_at - - now = time.time() - if now - _admob_keys_fetched_at > _ADMOB_KEY_TTL or not _admob_key_cache: - try: - resp = requests.get(ADMOB_KEYS_URL, timeout=5) - resp.raise_for_status() - data = resp.json() - - new_cache = {} - for entry in data.get("keys", []): - kid = str(entry["keyId"]) - pem = entry["pem"].encode() - new_cache[kid] = load_pem_public_key(pem) - - _admob_key_cache = new_cache - _admob_keys_fetched_at = now - print(f"[ADMOB] Refreshed {len(new_cache)} public key(s) from Google.") - - except Exception as e: - print(f"[ADMOB] WARNING: Could not refresh public keys: {e}") - # Fall through with stale cache rather than rejecting all valid callbacks - - return _admob_key_cache.get(key_id) - - -def _verify_admob_signature(raw_query_string: str, signature_b64url: str, key_id: str) -> bool: - """ - Verifies an AdMob SSV ECDSA-SHA256 signature. - Google signs all query params EXCEPT 'signature' and 'key_id', in original order. - """ - try: - # 1. Rebuild the signed message: all params except signature & key_id - params = urllib.parse.parse_qsl(raw_query_string, keep_blank_values=True) - filtered = [(k, v) for k, v in params if k not in ("signature", "key_id")] - message = urllib.parse.urlencode(filtered).encode("utf-8") - - # 2. Base64url-decode the signature (add missing padding if needed) - padding = 4 - len(signature_b64url) % 4 - sig_bytes = base64.urlsafe_b64decode(signature_b64url + ("=" * (padding % 4))) - - # 3. Fetch the matching public key - pub_key = _get_admob_public_key(key_id) - if pub_key is None: - print(f"[ADMOB] Unknown key_id: {key_id}. Rejecting.") - return False - - # 4. Verify — raises InvalidSignature on failure - pub_key.verify(sig_bytes, message, ec.ECDSA(hashes.SHA256())) - return True - - except Exception as e: - print(f"[ADMOB] Signature verification failed: {e}") - return False - - -@app.route('/api/admob/callback', methods=['GET']) -def admob_callback(): - """ - Google AdMob SSV callback — called server-to-server when a rewarded ad completes. - The Android client never touches this endpoint; Google calls it directly. - - custom_data (JSON, set by the app before showing the ad): - { "device_id": "...", "public_key": "...", "ad_type": "main_ad"|"bonus_ad" } - """ - try: - # ====================================================================== - # ⚠️ ADMOB BYPASS (TEMPORARY FOR APP REVIEW) - # DELETE THIS ENTIRE BLOCK WHEN YOU WANT TO ENABLE REAL ADS - # ====================================================================== - custom_data_str = request.args.get('custom_data') - if not custom_data_str: - return "OK", 200 - # ====================================================================== - - # ====================================================================== - # ✅ REAL ADMOB CODE - # UNCOMMENT THIS ENTIRE BLOCK WHEN YOU WANT TO ENABLE REAL ADS - # ====================================================================== - # signature = request.args.get('signature') - # key_id = request.args.get('key_id') - # custom_data_str = request.args.get('custom_data') - # - # # Google sends a verification ping with signature+key_id but no custom_data - # if not custom_data_str: - # return "OK", 200 # Verification ping from Google dashboard - # - # if not all([signature, key_id]): - # return "Missing parameters", 400 - # - # raw_qs = request.query_string.decode("utf-8") - # if not _verify_admob_signature(raw_qs, signature, key_id): - # print("[ADMOB] Rejected callback — invalid signature.") - # return "Forbidden", 403 - # ====================================================================== - - custom_data = json.loads(custom_data_str) - device_id = custom_data.get('device_id') - client_pubkey = custom_data.get('public_key') - - if not device_id or not client_pubkey: - return "Invalid custom_data payload", 400 - - ad_type = custom_data.get('ad_type', 'main_ad') - - success, msg = _start_or_extend_session(device_id, client_pubkey, ad_type) - return ("OK", 200) if success else (msg, 500) - - except Exception as e: - print(f"[ADMOB] SSV Error: {e}") - return "Internal Error", 500 - - -# ============================================================================== -# FLASK API (APP POLLING) -# ============================================================================== - -@app.route('/api/session/status', methods=['GET']) -def session_status(): - """ - Query: ?device_id=esf-xxx - Returns current session state so the Android app can construct its AmneziaWG config. - """ - device_id = request.args.get('device_id') - if not device_id: - return jsonify({"error": "Missing query parameter: device_id"}), 400 - - with sessions_lock: - session = active_sessions.get(device_id) - - if not session: - return jsonify({"active": False, "message": "No session found."}), 404 - - # Bulletproof fallback: If server public key is missing, try fetching it right now - global SERVER_PUBKEY - if not SERVER_PUBKEY: - pubkey = run_cmd("awg show awg0 public-key") - if pubkey: - SERVER_PUBKEY = pubkey - print(f"[AWG] Dynamic Server Public Key cached from awg0: {SERVER_PUBKEY}") - else: - SERVER_PUBKEY = SERVER_PUBKEY_FALLBACK or "server key=" - - stats = get_wireguard_stats() - current_bytes = stats.get(session["ip"], {}).get("total_bytes", 0) - used_bytes = max(0, current_bytes - session["baseline_bytes"]) - remaining_data = max(0, session["quota_bytes"] - used_bytes) - remaining_secs = max(0, session["expires_at"] - time.time()) - - return jsonify({ - "active": remaining_secs > 0, - "client_ip": session["ip"], - "client_ipv6": session.get("ipv6", ""), - "server_pubkey": SERVER_PUBKEY, - "is_throttled": session["is_throttled"], - "expires_in_seconds": int(remaining_secs), - "quota_bytes": session["quota_bytes"], - "used_bytes": used_bytes, - "remaining_bytes": remaining_data, - }) - - -@app.route('/api/session/stop', methods=['POST']) -def session_stop(): - """Allows the app to explicitly tear down its session on disconnect.""" - data = request.get_json(silent=True) - device_id = data.get('device_id') if data else None - - if device_id: - with sessions_lock: - session = active_sessions.pop(device_id, None) - if session: - remove_wg_peer(session["pubkey"]) - if session["is_throttled"]: - remove_throttle(session["ip"]) - release_ip(session["ip"]) - print(f"[API] Explicit stop for {device_id}.") - - return jsonify({"success": True}) - - # ============================================================================== # ENTRY POINT # ==============================================================================