gateway: protect external API access

common/etc/eterban/settings.ini

@@ -37,6 +37,11 @@
 # API server for checking if IP is banned
 listen_host = 127.0.0.1
 listen_port = 8275
+# Required when listen_host is not a literal loopback address. Send it as:
+# Authorization: Bearer <api_token>
+#api_token = change-this-to-a-long-random-secret
+# Per-client request limit. Default: 60.
+#rate_limit_per_minute = 60
 
 [AutoUnban]
 # Enable automatic unbanning with exponential backoff

gateway/usr/share/eterban/eterban_api.py

@@ -6,8 +6,12 @@ import json
 import subprocess
 import ipaddress
 import configparser
+import hmac
 import os
 import sys
+import threading
+import time
+from collections import defaultdict, deque
 
 LISTEN_HOST = '127.0.0.1'
 LISTEN_PORT = 8275
@@ -19,6 +23,18 @@ IPSET_WHITE = 'eterban_white'
 IPSET_WHITE_V6 = 'eterban_white_ipv6'
 
 path_to_config = '/etc/eterban/settings.ini'
+API_TOKEN = ''
+API_RATE_LIMIT = 60
+request_times = defaultdict(deque)
+request_lock = threading.Lock()
+
+
+def is_loopback_host(host):
+    """Return True only for literal loopback addresses."""
+    try:
+        return ipaddress.ip_address(host).is_loopback
+    except ValueError:
+        return False
 
 
 def ipset_test(setname, ip):
@@ -64,36 +80,58 @@ def check_ip(ip_str):
 
 
 class EterbanAPIHandler(http.server.BaseHTTPRequestHandler):
+    def _send_json(self, status, result):
+        self.send_response(status)
+        self.send_header('Content-Type', 'application/json')
+        self.end_headers()
+        self.wfile.write(json.dumps(result).encode())
+
+    def _authorized(self):
+        if not API_TOKEN:
+            return True
+        header = self.headers.get('Authorization', '')
+        return header.startswith('Bearer ') and hmac.compare_digest(header[7:], API_TOKEN)
+
+    def _rate_limited(self):
+        now = time.monotonic()
+        client = self.client_address[0]
+        with request_lock:
+            timestamps = request_times[client]
+            while timestamps and timestamps[0] <= now - 60:
+                timestamps.popleft()
+            if len(timestamps) >= API_RATE_LIMIT:
+                return True
+            timestamps.append(now)
+        return False
+
     def do_GET(self):
+        if self._rate_limited():
+            self._send_json(429, {'error': 'rate limit exceeded'})
+            return
+        if not self._authorized():
+            self._send_json(401, {'error': 'authentication required'})
+            return
         # /check/<ip>
         if self.path.startswith('/check/'):
             ip_str = self.path[len('/check/'):]
             result = check_ip(ip_str)
             status = 400 if 'error' in result else 200
-            self.send_response(status)
-            self.send_header('Content-Type', 'application/json')
-            self.end_headers()
-            self.wfile.write(json.dumps(result).encode())
+            self._send_json(status, result)
             return
 
         # /health
         if self.path == '/health':
-            self.send_response(200)
-            self.send_header('Content-Type', 'application/json')
-            self.end_headers()
-            self.wfile.write(json.dumps({'status': 'ok'}).encode())
+            self._send_json(200, {'status': 'ok'})
             return
 
-        self.send_response(404)
-        self.send_header('Content-Type', 'application/json')
-        self.end_headers()
-        self.wfile.write(json.dumps({'error': 'not found', 'usage': '/check/<ip>'}).encode())
+        self._send_json(404, {'error': 'not found', 'usage': '/check/<ip>'})
 
     def log_message(self, format, *args):
-        pass  # suppress default logging
+        sys.stderr.write('%s - %s\n' % (self.client_address[0], format % args))
 
 
 def main():
+    global API_TOKEN, API_RATE_LIMIT
     host = LISTEN_HOST
     port = LISTEN_PORT
 
@@ -103,6 +141,13 @@ def main():
         config.read(path_to_config)
         host = config.get('API', 'listen_host', fallback=LISTEN_HOST)
         port = config.getint('API', 'listen_port', fallback=LISTEN_PORT)
+        API_TOKEN = config.get('API', 'api_token', fallback='').strip()
+        API_RATE_LIMIT = config.getint('API', 'rate_limit_per_minute', fallback=60)
+
+    if not is_loopback_host(host) and not API_TOKEN:
+        raise RuntimeError('API token is required for a non-loopback listen_host')
+    if API_RATE_LIMIT < 1:
+        raise RuntimeError('API rate_limit_per_minute must be positive')
 
     server = http.server.HTTPServer((host, port), EterbanAPIHandler)
     print(f"Eterban API listening on {host}:{port}")