diff --git a/swarm_provenance_mcp/gateway_client.py b/swarm_provenance_mcp/gateway_client.py index b884342..1cbdb65 100644 --- a/swarm_provenance_mcp/gateway_client.py +++ b/swarm_provenance_mcp/gateway_client.py @@ -141,7 +141,7 @@ def extend_stamp(self, stamp_id: str, duration_hours: int) -> Dict[str, Any]: return response.json() def upload_data( - self, data: str, stamp_id: str, content_type: str = "application/json" + self, data: str, stamp_id: str, content_type: str = "application/json", sign: Optional[str] = None ) -> Dict[str, Any]: """Upload data to Swarm network. @@ -149,6 +149,8 @@ def upload_data( data: Data content as string (max 4096 bytes) stamp_id: Postage stamp ID to use for upload content_type: MIME type of the content (default: application/json) + sign: Signing method for provenance. Use 'notary' to have the gateway + cryptographically sign the data at upload time. Returns: Upload response with reference hash @@ -170,6 +172,8 @@ def upload_data( files = {"file": ("data", data_bytes, content_type)} params = {"stamp_id": stamp_id, "content_type": content_type} + if sign: + params["sign"] = sign # For file uploads, temporarily remove Content-Type from session to let requests set multipart/form-data original_content_type = self.session.headers.pop("Content-Type", None) diff --git a/swarm_provenance_mcp/server.py b/swarm_provenance_mcp/server.py index d71701a..cbd3c2c 100644 --- a/swarm_provenance_mcp/server.py +++ b/swarm_provenance_mcp/server.py @@ -582,6 +582,11 @@ async def list_tools() -> List[Tool]: "description": "MIME type of the content (e.g., application/json, text/plain, image/png)", "default": "application/json", }, + "sign": { + "type": "string", + "description": "Signing method for provenance. Use 'notary' to have the gateway cryptographically sign the data at upload time, creating a verifiable proof of upload. Check notary_info first to confirm signing is available. Leave empty for unsigned upload.", + "enum": ["notary"], + }, }, "required": ["data", "stamp_id"], }, @@ -1505,6 +1510,7 @@ async def handle_upload_data(arguments: Dict[str, Any]) -> CallToolResult: if not stamp_id: raise ValueError("Stamp ID cannot be empty") content_type = arguments.get("content_type", "application/json") + sign = arguments.get("sign") # Validate inputs validate_data_size(data) @@ -1553,17 +1559,26 @@ async def handle_upload_data(arguments: Dict[str, Any]) -> CallToolResult: raise # Proceed with upload if stamp validation passed - result = gateway_client.upload_data(data, clean_stamp_id, content_type) + result = gateway_client.upload_data(data, clean_stamp_id, content_type, sign=sign) response_text = f"šŸŽ‰ Data uploaded successfully to Swarm!\n\n" response_text += f"šŸ“„ Upload Details:\n" response_text += f" Size: {len(data.encode('utf-8')):,} bytes\n" response_text += f" Content Type: {content_type}\n" - response_text += f" Stamp Used: `{clean_stamp_id}`\n\n" - response_text += f"šŸ”— Retrieval Information:\n" + response_text += f" Stamp Used: `{clean_stamp_id}`\n" + if sign: + response_text += f" Signing: {sign}\n" + response_text += f"\nšŸ”— Retrieval Information:\n" response_text += f" Reference Hash: `{result['reference']}`\n" response_text += f" šŸ’” Copy this reference hash to download your data later using the 'download_data' tool." + # Surface notary signature details when present + notary_info = result.get("notary") or result.get("signature") + if notary_info and isinstance(notary_info, dict): + signer = notary_info.get("signer") or notary_info.get("signerAddress") + if signer: + response_text += f"\n\nšŸ” Notary Signature:\n Signer: {signer}" + # Add validation warning if applicable if stamp_validation_failed: response_text += f"\nNote: {validation_error_msg}" diff --git a/tests/test_gateway_client.py b/tests/test_gateway_client.py index ce87ce4..c008f7b 100644 --- a/tests/test_gateway_client.py +++ b/tests/test_gateway_client.py @@ -179,3 +179,31 @@ def test_custom_headers(self): assert "swarm-provenance-mcp" in request_headers["User-Agent"] assert "X-Payment-Mode" in request_headers assert request_headers["X-Payment-Mode"] == "free" + + def test_upload_data_unsigned(self): + """Test that unsigned upload sends no sign query param.""" + with requests_mock.Mocker() as m: + stamp_id = "a" * 64 + expected_response = {"reference": "b" * 64} + m.post(f"{self.base_url}/api/v1/data/", json=expected_response) + + result = self.client.upload_data('{"key": "value"}', stamp_id) + + assert result["reference"] == "b" * 64 + assert "sign" not in m.last_request.qs + + def test_upload_data_with_notary_sign(self): + """Test that notary sign upload includes ?sign=notary query param.""" + with requests_mock.Mocker() as m: + stamp_id = "a" * 64 + signer_address = "0xabcdef1234567890abcdef1234567890abcdef12" + expected_response = { + "reference": "c" * 64, + "notary": {"signer": signer_address}, + } + m.post(f"{self.base_url}/api/v1/data/", json=expected_response) + + result = self.client.upload_data('{"key": "value"}', stamp_id, sign="notary") + + assert result["reference"] == "c" * 64 + assert m.last_request.qs.get("sign") == ["notary"]