diff --git a/.gitea/workflows/deploy-my-vpn.yml b/.gitea/workflows/deploy-my-vpn.yml index 4e40c7b..3ec4c21 100644 --- a/.gitea/workflows/deploy-my-vpn.yml +++ b/.gitea/workflows/deploy-my-vpn.yml @@ -9,6 +9,13 @@ jobs: deploy: runs-on: ubuntu-latest steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Unit tests + run: | + python3 -m unittest discover -s tests + - name: Deploy via SSH env: VPN_HOST: ${{ secrets.VPN_HOST }} diff --git a/app.py b/app.py index f53e229..5944a97 100644 --- a/app.py +++ b/app.py @@ -5,20 +5,32 @@ from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, HTTPServer +def build_payload(headers, client_ip, method, path, now=None): + if now is None: + now = datetime.now(timezone.utc) + + forwarded = headers.get("X-Forwarded-For", "") + ip = forwarded.split(",")[0].strip() if forwarded else client_ip + + return { + "ip": ip, + "user_agent": headers.get("User-Agent", ""), + "method": method, + "path": path, + "timestamp": now.isoformat(), + "headers": dict(headers), + } + + class Handler(BaseHTTPRequestHandler): def _write_json(self, status=200): - client_ip = self.headers.get("X-Forwarded-For", "").split(",")[0].strip() - if not client_ip: - client_ip = self.client_address[0] - - payload = { - "ip": client_ip, - "user_agent": self.headers.get("User-Agent", ""), - "method": self.command, - "path": self.path, - "timestamp": datetime.now(timezone.utc).isoformat(), - "headers": {k: v for k, v in self.headers.items()}, - } + headers = {k: v for k, v in self.headers.items()} + payload = build_payload( + headers=headers, + client_ip=self.client_address[0], + method=self.command, + path=self.path, + ) data = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) diff --git a/tests/test_app.py b/tests/test_app.py new file mode 100644 index 0000000..04ffd5f --- /dev/null +++ b/tests/test_app.py @@ -0,0 +1,40 @@ +import unittest +from datetime import datetime, timezone + +from app import build_payload + + +class TestBuildPayload(unittest.TestCase): + def test_uses_forwarded_ip(self): + now = datetime(2026, 2, 6, 0, 0, 0, tzinfo=timezone.utc) + payload = build_payload( + headers={"X-Forwarded-For": "10.0.0.1, 10.0.0.2", "User-Agent": "ua"}, + client_ip="192.168.0.10", + method="GET", + path="/", + now=now, + ) + self.assertEqual(payload["ip"], "10.0.0.1") + self.assertEqual(payload["user_agent"], "ua") + self.assertEqual(payload["method"], "GET") + self.assertEqual(payload["path"], "/") + self.assertEqual(payload["timestamp"], "2026-02-06T00:00:00+00:00") + + def test_falls_back_to_client_ip(self): + now = datetime(2026, 2, 6, 0, 0, 0, tzinfo=timezone.utc) + payload = build_payload( + headers={"User-Agent": "ua"}, + client_ip="192.168.0.10", + method="POST", + path="/submit", + now=now, + ) + self.assertEqual(payload["ip"], "192.168.0.10") + self.assertEqual(payload["user_agent"], "ua") + self.assertEqual(payload["method"], "POST") + self.assertEqual(payload["path"], "/submit") + self.assertEqual(payload["timestamp"], "2026-02-06T00:00:00+00:00") + + +if __name__ == "__main__": + unittest.main()