test: cover agent pipe transfer validation

agent-pipe-uploader/Containerfile

@@ -11,6 +11,10 @@ RUN pip install --no-cache-dir "fastmcp==${FASTMCP_VERSION}" \
 
 WORKDIR /app
 COPY server.py /app/server.py
+COPY test_server.py /tmp/test_server.py
+
+RUN PYTHONPATH=/app python3 /tmp/test_server.py \
+    && rm -f /tmp/test_server.py
 
 USER 1000
 

agent-pipe-uploader/README.md

@@ -8,3 +8,6 @@ The chart mounts a non-secret profile file at
 `/etc/agent-pipe/profiles.json` and an isolated artifact PVC at `/artifacts`.
 The service exposes Streamable HTTP MCP at `/mcp` and health at `/healthz`.
 It does not log signed URLs or artifact bytes.
+
+`test_server.py` is run during the image build and covers artifact-root escape
+rejection plus signed-URL host, prefix, and authorization-parameter validation.

agent-pipe-uploader/test_server.py

@@ -0,0 +1,77 @@
+import importlib
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+
+class ServerTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.temporary_directory = tempfile.TemporaryDirectory()
+        cls.root = Path(cls.temporary_directory.name)
+        cls.artifacts = cls.root / "artifacts"
+        cls.artifacts.mkdir()
+        cls.profiles = cls.root / "profiles.json"
+        cls.profiles.write_text(
+            json.dumps(
+                {
+                    "profiles": {
+                        "agent-pipe": {
+                            "allowedHosts": ["agent-pipe.s3.us-west-2.amazonaws.com"],
+                            "pathPrefixes": ["/deliveries/"],
+                            "requiredQueryParameters": ["X-Amz-Algorithm", "X-Amz-Signature"],
+                            "maxBytes": 1024,
+                        }
+                    }
+                }
+            )
+        )
+        os.environ["ARTIFACT_ROOT"] = str(cls.artifacts)
+        os.environ["PROFILE_CONFIG_PATH"] = str(cls.profiles)
+        os.environ["MCP_ALLOWED_HOSTS"] = "agent-pipe-uploader.opencode.svc"
+        cls.server = importlib.import_module("server")
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.temporary_directory.cleanup()
+
+    def test_resolves_existing_artifact(self):
+        artifact = self.artifacts / "sample.txt"
+        artifact.write_text("test")
+        self.assertEqual(self.server.artifact_path("sample.txt", must_exist=True), artifact)
+
+    def test_rejects_artifact_path_escape(self):
+        with self.assertRaises(self.server.TransferError):
+            self.server.artifact_path("../outside", must_exist=True)
+
+    def test_validates_profile_signed_url(self):
+        profile, host, target = self.server.signed_target(
+            "agent-pipe",
+            "https://agent-pipe.s3.us-west-2.amazonaws.com/deliveries/test.txt?"
+            "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Signature=example",
+        )
+        self.assertEqual(profile["maxBytes"], 1024)
+        self.assertEqual(host, "agent-pipe.s3.us-west-2.amazonaws.com")
+        self.assertTrue(target.startswith("/deliveries/test.txt?"))
+
+    def test_rejects_unapproved_signed_url(self):
+        with self.assertRaises(self.server.TransferError):
+            self.server.signed_target(
+                "agent-pipe",
+                "https://example.com/deliveries/test.txt?"
+                "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Signature=example",
+            )
+
+    def test_rejects_disallowed_object_prefix(self):
+        with self.assertRaises(self.server.TransferError):
+            self.server.signed_target(
+                "agent-pipe",
+                "https://agent-pipe.s3.us-west-2.amazonaws.com/private/test.txt?"
+                "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Signature=example",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()