|
| 1 | +""" |
| 2 | +RelayListener |
| 3 | +Receives worklist actions from manage-screening. |
| 4 | +Supports creation of Modality Worklist Items. |
| 5 | +""" |
| 6 | + |
| 7 | +import asyncio |
| 8 | +import base64 |
| 9 | +import hashlib |
| 10 | +import hmac |
| 11 | +import json |
| 12 | +import logging |
| 13 | +import os |
| 14 | +import time |
| 15 | +import urllib.parse |
| 16 | + |
| 17 | +from websockets.asyncio.client import connect |
| 18 | + |
| 19 | +from services.mwl.create_worklist_item import CreateWorklistItem |
| 20 | +from services.storage import MWLStorage |
| 21 | + |
| 22 | +logger = logging.getLogger(__name__) |
| 23 | + |
| 24 | +DB_PATH = os.getenv("MWL_DB_PATH", "/var/lib/pacs/worklist.db") |
| 25 | +SAS_TOKEN_EXPIRY_SECONDS = 3600 |
| 26 | + |
| 27 | +ACTIONS = { |
| 28 | + "worklist.create_item": CreateWorklistItem, |
| 29 | +} |
| 30 | + |
| 31 | + |
| 32 | +class RelayListener: |
| 33 | + """Socket Listener for Azure Relay.""" |
| 34 | + |
| 35 | + def __init__(self, storage: MWLStorage): |
| 36 | + self.storage = storage |
| 37 | + self.relay_uri = RelayURI() |
| 38 | + |
| 39 | + async def listen(self): |
| 40 | + """Listen for messages from Azure Relay.""" |
| 41 | + |
| 42 | + logger.info(f"Connecting to Azure Relay: {self.relay_uri.hybrid_connection_name}...") |
| 43 | + |
| 44 | + async with connect(self.relay_uri.connection_url(), compression=None) as websocket: |
| 45 | + logger.info("Connected - waiting for worklist actions...") |
| 46 | + |
| 47 | + async for message in websocket: |
| 48 | + try: |
| 49 | + data = json.loads(message) |
| 50 | + |
| 51 | + if "accept" in data: |
| 52 | + accept_url = data["accept"]["address"] |
| 53 | + logger.info("Incoming connection...") |
| 54 | + |
| 55 | + async with connect(accept_url, compression=None) as client_ws: |
| 56 | + client_message = await asyncio.wait_for(client_ws.recv(), timeout=30) |
| 57 | + payload = json.loads(client_message) |
| 58 | + response = self.process_action(payload) |
| 59 | + |
| 60 | + # Send acknowledgment |
| 61 | + await client_ws.send(json.dumps(response)) |
| 62 | + |
| 63 | + except asyncio.TimeoutError: |
| 64 | + logger.error("Timeout waiting for message") |
| 65 | + except Exception as e: |
| 66 | + logger.error(f"Error: {e}") |
| 67 | + |
| 68 | + def process_action(self, payload: dict): |
| 69 | + """Process incoming action payload.""" |
| 70 | + action_name = payload.get("action_type", "no-op") |
| 71 | + |
| 72 | + action_class = ACTIONS.get(action_name) |
| 73 | + if not action_class: |
| 74 | + raise ValueError(f"Unknown action: {action_name}") |
| 75 | + |
| 76 | + return action_class(self.storage).call(payload) |
| 77 | + |
| 78 | + |
| 79 | +class RelayURI: |
| 80 | + def __init__(self): |
| 81 | + self.relay_namespace = os.getenv("AZURE_RELAY_NAMESPACE", "relay-test.servicebus.windows.net") |
| 82 | + self.hybrid_connection_name = os.getenv("AZURE_RELAY_HYBRID_CONNECTION", "relay-test-hc") |
| 83 | + self.key_name = os.getenv("AZURE_RELAY_KEY_NAME", "RootManageSharedAccessKey") |
| 84 | + self.shared_access_key = os.getenv("AZURE_RELAY_SHARED_ACCESS_KEY", "") |
| 85 | + |
| 86 | + def create_sas_token(self, expiry_seconds: int = SAS_TOKEN_EXPIRY_SECONDS) -> str: |
| 87 | + """Create SAS token for Azure Relay authentication.""" |
| 88 | + uri = f"http://{self.relay_namespace}/{self.hybrid_connection_name}" |
| 89 | + encoded_uri = urllib.parse.quote_plus(uri) |
| 90 | + expiry = str(int(time.time() + expiry_seconds)) |
| 91 | + signature = base64.b64encode( |
| 92 | + hmac.new(self.shared_access_key.encode(), f"{encoded_uri}\n{expiry}".encode(), hashlib.sha256).digest() |
| 93 | + ) |
| 94 | + return ( |
| 95 | + f"SharedAccessSignature sr={encoded_uri}" |
| 96 | + f"&sig={urllib.parse.quote_plus(signature)}" |
| 97 | + f"&se={expiry}&skn={self.key_name}" |
| 98 | + ) |
| 99 | + |
| 100 | + def connection_url(self) -> str: |
| 101 | + token = self.create_sas_token() |
| 102 | + return ( |
| 103 | + f"wss://{self.relay_namespace}/$hc/{self.hybrid_connection_name}" |
| 104 | + f"?sb-hc-action=listen&sb-hc-token={urllib.parse.quote_plus(token)}" |
| 105 | + ) |
| 106 | + |
| 107 | + |
| 108 | +async def main(): |
| 109 | + logging.basicConfig( |
| 110 | + level=os.getenv("LOG_LEVEL", "INFO").upper(), |
| 111 | + format=os.getenv("LOG_FORMAT", "%(asctime)s - %(name)s - %(levelname)s - %(message)s"), |
| 112 | + ) |
| 113 | + |
| 114 | + logger.info("Socket Listener Starting...") |
| 115 | + storage = MWLStorage(db_path=DB_PATH) |
| 116 | + |
| 117 | + while True: |
| 118 | + try: |
| 119 | + await RelayListener(storage).listen() |
| 120 | + except KeyboardInterrupt: |
| 121 | + logger.warning("\nShutting down...") |
| 122 | + break |
| 123 | + except Exception as e: |
| 124 | + logger.warning(f"Connection error: {e}") |
| 125 | + logger.warning("Retrying in 5 seconds...") |
| 126 | + await asyncio.sleep(5) |
| 127 | + |
| 128 | + |
| 129 | +if __name__ == "__main__": |
| 130 | + asyncio.run(main()) |
0 commit comments