Add backend hardening tests

tests/test_hardening.py

@@ -0,0 +1,116 @@
+import threading
+from time import monotonic
+
+import httpx
+from flask import Flask
+
+from upload.backend.session import add_file, cleanup, configure, create_session, get_files
+from upload.backend.session import state as session_state
+from upload.backend.upload_refs.blueprint import create_upload_refs_blueprint
+
+
+def build_app(transport, **config):
+    app = Flask(__name__)
+    app.config['TESTING'] = True
+    app.config['UPLOAD_HTTPX_TRANSPORT'] = transport
+    app.register_blueprint(create_upload_refs_blueprint({
+        'vmUploadPrefix': 'https://vm.test/buffer/',
+        'httpxTransport': transport,
+        'pullRetries': 3,
+        'pullRetryDelay': 0,
+        **config,
+    }))
+    return app
+
+
+def test_session_limit_accepts_exact_boundary_and_rejects_overflow():
+    sid = create_session()
+    previous = session_state.MAX_SESSION_BYTES
+    configure(max_session_bytes=10)
+    try:
+        assert add_file(sid, 'exact.bin', b'x' * 10) is True
+        assert add_file(sid, 'overflow.bin', b'x') is False
+        assert len(get_files(sid)) == 1
+    finally:
+        configure(max_session_bytes=previous)
+        cleanup(sid)
+
+
+def test_upload_refs_rejects_actual_oversize_and_deletes_buffer():
+    storage = {'oversize': b'123456789'}
+    deleted = []
+
+    def handler(request):
+        if request.method == 'GET':
+            return httpx.Response(200, content=storage['oversize'])
+        if request.method == 'DELETE':
+            deleted.append(request.url.path)
+            storage.clear()
+            return httpx.Response(204)
+        return httpx.Response(405)
+
+    transport = httpx.MockTransport(handler)
+    app = build_app(transport, maxFileBytes=8)
+    sid = create_session()
+    try:
+        with app.test_client() as client:
+            response = client.post('/api/upload_refs', json={
+                'session': sid,
+                'files': [{
+                    'name': 'oversize.txt',
+                    'size': 8,
+                    'url': 'https://vm.test/buffer/oversize',
+                }],
+            })
+        assert response.status_code == 200
+        assert response.get_json()['added'] == 0
+        assert get_files(sid) == []
+        assert deleted == ['/buffer/oversize']
+        assert storage == {}
+    finally:
+        cleanup(sid)
+
+
+def test_upload_refs_retries_transient_pull_and_deletes_after_success():
+    attempts = []
+    deleted = []
+
+    def handler(request):
+        if request.method == 'GET':
+            attempts.append(request.url.path)
+            if len(attempts) == 1:
+                return httpx.Response(503)
+            return httpx.Response(200, content=b'retry-content')
+        if request.method == 'DELETE':
+            deleted.append(request.url.path)
+            return httpx.Response(204)
+        return httpx.Response(405)
+
+    transport = httpx.MockTransport(handler)
+    app = build_app(transport)
+    with app.test_client() as client:
+        response = client.post('/api/upload_refs', json={'files': [{
+            'name': 'retry.txt',
+            'size': 13,
+            'url': 'https://vm.test/buffer/retry',
+        }]})
+    assert response.status_code == 200
+    data = response.get_json()
+    assert data['added'] == 1
+    assert len(attempts) == 2
+    assert deleted == ['/buffer/retry']
+    cleanup(data['session'])
+
+
+def test_session_ttl_removes_session():
+    previous = session_state.TTL_SECONDS
+    configure(ttl_seconds=0.05)
+    sid = create_session()
+    try:
+        deadline = monotonic() + 1
+        while monotonic() < deadline and get_files(sid) is not None:
+            threading.Event().wait(0.01)
+        assert get_files(sid) is None
+    finally:
+        configure(ttl_seconds=previous)
+        cleanup(sid)