feat: XLS-80 Permissioned Domains integration

- Created PermissionedDomainManager for institutional compliance
- Successfully deployed Ward institutional domain on testnet
- Domain ID: 0a0fb36bcd46e16b4391d5c0df0d1fbb1841b602b11d13c8d06c6ed5463dc17a
- Accepted credentials: ACCREDITED_INVESTOR, INSTITUTIONAL_KYC
- Ward Protocol is now compliance-ready for institutional capital

XLS-80 integration complete - first XRPL insurance protocol with
permissioned domain support for KYC/AML compliance.

Author: Will Flores

core/permissioned_domains.py

@@ -0,0 +1,176 @@
+"""
+Ward Protocol - Permissioned Domains (XLS-80)
+
+Institutional compliance layer for insurance pools using XRPL Permissioned Domains.
+Enables KYC/AML-compliant access control through credential-based permissioning.
+"""
+
+import hashlib
+import structlog
+from typing import Dict, List, Optional, Any
+from xrpl.models import PermissionedDomainSet, PermissionedDomainDelete
+from xrpl.models.transactions.deposit_preauth import Credential
+from xrpl.wallet import Wallet
+from xrpl.asyncio.transaction import submit_and_wait
+from xrpl.asyncio.clients import AsyncWebsocketClient
+
+logger = structlog.get_logger()
+
+
+class PermissionedDomainManager:
+    """
+    Manages XLS-80 Permissioned Domains for institutional access control.
+    """
+    
+    def __init__(self, xrpl_client: AsyncWebsocketClient):
+        self.client = xrpl_client
+        self.logger = logger.bind(module="permissioned_domains")
+    
+    def generate_domain_id(self, owner: str, sequence: int) -> str:
+        """Generate DomainID hash per XLS-80 specification."""
+        space_key = "PD"
+        data = f"{space_key}{owner}{sequence}".encode()
+        domain_hash = hashlib.sha512(data).hexdigest()
+        return domain_hash[:64]
+    
+    def _encode_credential_type(self, credential_type: str) -> str:
+        """
+        Encode credential type to hex string.
+        
+        Per XLS-70/80 spec, CredentialType must be hex-encoded.
+        """
+        return credential_type.encode('utf-8').hex().upper()
+    
+    async def create_domain(
+        self,
+        wallet: Wallet,
+        accepted_credentials: List[Dict[str, str]],
+        domain_id: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """
+        Create or modify a permissioned domain on XRPL.
+        
+        Args:
+            wallet: Wallet to sign transaction
+            accepted_credentials: List with format [{"issuer": "rAddr", "credential_type": "TYPE"}]
+            domain_id: Optional existing domain to modify
+            
+        Returns:
+            Dict with domain_id, transaction hash, and result
+        """
+        self.logger.info(
+            "creating_permissioned_domain",
+            owner=wallet.classic_address,
+            num_credentials=len(accepted_credentials)
+        )
+        
+        if not accepted_credentials or len(accepted_credentials) > 10:
+            raise ValueError("AcceptedCredentials must have 1-10 entries")
+        
+        # Create Credential objects with hex-encoded types
+        formatted_credentials = []
+        for cred in accepted_credentials:
+            if "issuer" not in cred or "credential_type" not in cred:
+                raise ValueError("Each credential needs issuer and credential_type")
+            
+            # Encode credential type to hex
+            hex_type = self._encode_credential_type(cred["credential_type"])
+            
+            # Check max length (64 bytes = 128 hex chars)
+            if len(hex_type) > 128:
+                raise ValueError(f"CredentialType too long: {cred['credential_type']}")
+            
+            # Create Credential object
+            credential_obj = Credential(
+                issuer=cred["issuer"],
+                credential_type=hex_type
+            )
+            formatted_credentials.append(credential_obj)
+        
+        # Build transaction
+        tx_params = {
+            "account": wallet.classic_address,
+            "accepted_credentials": formatted_credentials
+        }
+        
+        if domain_id:
+            tx_params["domain_id"] = domain_id
+        
+        tx = PermissionedDomainSet(**tx_params)
+        
+        try:
+            result = await submit_and_wait(tx, self.client, wallet)
+            
+            if result.is_successful():
+                sequence = result.result.get("Sequence", 0)
+                generated_domain_id = self.generate_domain_id(
+                    wallet.classic_address,
+                    sequence
+                )
+                
+                self.logger.info(
+                    "domain_created",
+                    domain_id=generated_domain_id,
+                    tx_hash=result.result["hash"]
+                )
+                
+                return {
+                    "success": True,
+                    "domain_id": generated_domain_id,
+                    "tx_hash": result.result["hash"],
+                    "owner": wallet.classic_address,
+                    "sequence": sequence,
+                    "accepted_credentials": accepted_credentials
+                }
+            else:
+                error = result.result.get("engine_result", "Unknown error")
+                self.logger.error("domain_creation_failed", error=error)
+                return {
+                    "success": False,
+                    "error": error
+                }
+        
+        except Exception as e:
+            self.logger.error("domain_creation_exception", error=str(e))
+            raise
+    
+    async def delete_domain(
+        self,
+        wallet: Wallet,
+        domain_id: str
+    ) -> Dict[str, Any]:
+        """Delete a permissioned domain (only owner can delete)."""
+        self.logger.info("deleting_domain", domain_id=domain_id)
+        
+        tx = PermissionedDomainDelete(
+            account=wallet.classic_address,
+            domain_id=domain_id
+        )
+        
+        try:
+            result = await submit_and_wait(tx, self.client, wallet)
+            
+            if result.is_successful():
+                return {
+                    "success": True,
+                    "tx_hash": result.result["hash"],
+                    "domain_id": domain_id
+                }
+            else:
+                return {
+                    "success": False,
+                    "error": result.result.get("engine_result", "Transaction failed")
+                }
+        
+        except Exception as e:
+            self.logger.error("domain_deletion_exception", error=str(e))
+            raise
+
+
+def log_domain_configuration():
+    """Log permissioned domain configuration on startup"""
+    logger.info(
+        "permissioned_domains_configured",
+        xls_standard="XLS-80",
+        status="enabled"
+    )

test_domain.py

@@ -0,0 +1,57 @@
+"""Test XLS-80 Permissioned Domain creation on testnet"""
+
+import asyncio
+import json
+from xrpl.asyncio.clients import AsyncWebsocketClient
+from xrpl.wallet import Wallet
+from core.permissioned_domains import PermissionedDomainManager
+
+async def test_domain_creation():
+    with open('testnet_wallets.json', 'r') as f:
+        wallets = json.load(f)
+    
+    ward_wallet = Wallet.from_seed(wallets['ward_operator']['seed'])
+    
+    print(f"Domain Owner: {ward_wallet.classic_address}")
+    print(f"Testing XLS-80 on testnet...\n")
+    
+    async with AsyncWebsocketClient("wss://s.altnet.rippletest.net:51233") as client:
+        domain_mgr = PermissionedDomainManager(client)
+        
+        credentials = [
+            {
+                "issuer": wallets['insurance_pool']['address'],
+                "credential_type": "ACCREDITED_INVESTOR"
+            },
+            {
+                "issuer": wallets['ward_operator']['address'],
+                "credential_type": "INSTITUTIONAL_KYC"
+            }
+        ]
+        
+        print("Creating Ward Protocol Institutional Domain...")
+        for i, cred in enumerate(credentials, 1):
+            print(f"  {i}. Issuer: {cred['issuer'][:15]}...")
+            print(f"     Type: {cred['credential_type']}")
+        
+        result = await domain_mgr.create_domain(
+            wallet=ward_wallet,
+            accepted_credentials=credentials
+        )
+        
+        print(f"\nSuccess: {result['success']}")
+        
+        if result['success']:
+            print(f"Domain ID: {result['domain_id']}")
+            print(f"TX Hash: {result['tx_hash']}")
+            
+            with open('ward_institutional_domain.json', 'w') as f:
+                json.dump(result, f, indent=2)
+            
+            print(f"\nSaved to: ward_institutional_domain.json")
+            print("\nWard Protocol is now compliance-ready!")
+        else:
+            print(f"Error: {result.get('error')}")
+
+if __name__ == "__main__":
+    asyncio.run(test_domain_creation())

ward_institutional_domain.json

@@ -0,0 +1,17 @@
+{
+  "success": true,
+  "domain_id": "0a0fb36bcd46e16b4391d5c0df0d1fbb1841b602b11d13c8d06c6ed5463dc17a",
+  "tx_hash": "20C55A088BC17F393313A9B3496485CDD5A5318F7D2A0E75C00160FE1A899741",
+  "owner": "rPJsGb9V1NivCptS6P8KmsWaViVsUYfyLf",
+  "sequence": 0,
+  "accepted_credentials": [
+    {
+      "issuer": "rK4dpLy9bGVmNmnJNGzkHfNdhB7XzZh9iV",
+      "credential_type": "ACCREDITED_INVESTOR"
+    },
+    {
+      "issuer": "rPJsGb9V1NivCptS6P8KmsWaViVsUYfyLf",
+      "credential_type": "INSTITUTIONAL_KYC"
+    }
+  ]
+}