import os import unittest import sys from pathlib import Path # Import from ca_core code_path = Path(__file__).parents[2] / "ca_core" sys.path.insert(0, str(code_path)) from ca_core.crypto.zenroom_service_client import ZenroomServiceClient class TestZenroomServiceClientIntegration(unittest.TestCase): """Service integration: runs against a live RESTroom/Zenroom HTTP service. Enable by setting: ZENROOM_BASE_URL=http://localhost:3300 """ @unittest.skipUnless(os.getenv("ZENROOM_BASE_URL"), "No ZENROOM_BASE_URL set") def test_end_to_end_smoke_8_calls(self): """Hits exactly these 8 endpoints once each (in typical service logs): 1) Generate-a-keypair,-reading-identity-from-data 2) Generate-public-key 3) Encrypt-a-message-with-the-password 4) Decrypt-the-message-with-the-password 5) Encrypt-a-message-for-two-recipients-using-asymmetric-cryptography 6) Decrypt-a-message-for-two-recipients-using-asymmetric-cryptography 7) Sign-objects-using-asymmetric-cryptography 8) Verify-asymmetric-cryptography-signature Note: To keep it to 8 calls, we reuse the same identity as both sender and receiver for the asymmetric encrypt/decrypt roundtrip. """ client = ZenroomServiceClient(base_url=os.environ["ZENROOM_BASE_URL"]) # 1) keypair (private) -> 2) public key sender_kp = client.generate_keypair("IntegrationUser123456") self.assertIn("keyring", sender_kp) self.assertIn("private_key", sender_kp) sender_pub = client.generate_public_key(sender_kp["keyring"]) self.assertIsInstance(sender_pub, str) self.assertTrue(sender_pub.strip()) # 3) symmetric encrypt -> 4) symmetric decrypt plaintext = "Dear Bob, your name is too short, goodbye - Alice." sm = client.symmetric_encrypt( header="A very important secret", message=plaintext, shared_key="myVerySecretPassword", ) pt = client.symmetric_decrypt(secret_message=sm, shared_key="myVerySecretPassword") self.assertEqual(pt, plaintext) # 5) asymmetric encrypt (receiver pub == sender pub to avoid extra keypair/public-key calls) secret = client.asymmetric_encrypt( receiver_public_key=sender_pub, sender_keyring=sender_kp["keyring"], message="Hello from integration test", header="Integration header", ) # 6) asymmetric decrypt (sender public key is sender_pub; receiver private key == sender private key) dec = client.asymmetric_decrypt( sender_public_key=sender_pub, receiver_keyring=sender_kp["keyring"], secret=secret, ) self.assertEqual(dec["header"], "Integration header") self.assertEqual(dec["text"], "Hello from integration test") # 7) sign -> 8) verify signed = client.sign_objects( signer_keyring=sender_kp["keyring"], objects={"myMessage": "Signed integration message"}, ) self.assertIn("myMessage.signature", signed) sig = signed["myMessage.signature"] self.assertIn("r", sig) self.assertIn("s", sig) ok = client.verify_signature( message_field="myMessage", message_value=signed["myMessage"], signature={"r": sig["r"], "s": sig["s"]}, signer_public_key=sender_pub, ) self.assertTrue(ok)