Patched code sample

import logging
import hmac
import hashlib
import json

# In a real-world scenario, this secret key would be securely stored, for example,
# in a hardware security module (HSM) or an encrypted configuration file, and not hardcoded.
# It must match the secret key used by authorized clients to sign their messages.
SHARED_SECRET_KEY = b'c3141592653589793238462643383279_a_very_secure_secret_key'

# Configure logging to show the outcome of the security checks.
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def write_and_bind_script(python_code, keybinding):
    """
    A placeholder function representing the vulnerable action of writing a script
    to disk and binding it to a controller key. In this fixed implementation, this
    function is only ever called after a request has been successfully authenticated.
    """
    # For safety, this example will not actually write to a sensitive system path.
    # It will log the action that would have been taken.
    logging.info(f"ACTION: Writing authenticated code to '/unitree/etc/programming/{keybinding}.py'")
    logging.info(f"ACTION: Binding script to keybinding '{keybinding}'")
    # Example of writing to a sandboxed location for review:
    # safe_path = f"/tmp/safe_scripts/{keybinding}.py"
    # with open(safe_path, "w") as f:
    #     f.write(python_code)
    pass

def process_programming_request(message):
    """
    Processes incoming DDS messages for the 'rt/api/programming_actuator/request' topic.
    This fixed version implements an authentication check using a message signature (HMAC)
    to fix the vulnerability described in CVE-2026-27509, which was caused by a
    lack of authentication.

    The message is now expected to have the following format:
    {
        "api_id": 1002,
        "payload": "<stringified_json_of_the_original_data>",
        "signature": "<hex_hmac_sha256_signature_of_the_payload_string>"
    }
    """
    api_id = message.get("api_id")
    
    if api_id != 1002:
        # This handler is only for the programmable actuator API endpoint.
        return

    logging.info("Received request for programmable actuator (api_id=1002). Verifying signature...")

    # --- START OF FIX ---
    # The vulnerability was a lack of authentication, allowing any network-adjacent
    # attacker to send commands. This fix adds a mandatory authentication layer
    # by requiring that the message payload be cryptographically signed with a
    # pre-shared secret key. This ensures message integrity and authenticity.
    # While the ideal fix is enabling native DDS-Security, this application-layer
    # check effectively mitigates the described unauthenticated access vector.
    
    try:
        payload_str = message.get("payload")
        received_signature = message.get("signature")

        if not payload_str or not received_signature:
            logging.warning("AUTHENTICATION FAILED: Request is malformed, missing 'payload' or 'signature'. Rejected.")
            return

        # 1. Recalculate the HMAC signature on the server side using the shared secret key.
        # The client must generate its signature using the exact same method and payload string.
        computed_signature = hmac.new(
            SHARED_SECRET_KEY,
            msg=payload_str.encode('utf-8'),
            digestmod=hashlib.sha256
        ).hexdigest()

        # 2. Securely compare the received signature with the one we just computed.
        # hmac.compare_digest is used to prevent timing attacks.
        if not hmac.compare_digest(computed_signature, received_signature):
            logging.warning("AUTHENTICATION FAILED: Invalid signature. The request may be tampered with or is from an unauthorized source. Rejected.")
            return

    except Exception as e:
        logging.error(f"An error occurred during the authentication process: {e}. Rejected.")
        return

    logging.info("AUTHENTICATION SUCCESSFUL. Proceeding to process the trusted request.")
    # --- END OF FIX ---

    # Now that authentication is complete, we can safely deserialize and process the payload.
    # It is still best practice to parse data inside a try-except block.
    try:
        data = json.loads(payload_str)
        python_code = data.get("code")
        keybinding = data.get("keybinding")

        if python_code and isinstance(python_code, str) and keybinding:
            # Defense-in-depth: Although authenticated, further sandboxing or validation
            # of 'python_code' (e.g., using an Abstract Syntax Tree to check for
            # forbidden modules like 'os' or 'subprocess') would be a powerful
            # secondary control. For this primary fix, we trust the authenticated source.
            write_and_bind_script(python_code, keybinding)
        else:
            logging.warning("Authenticated request is valid, but its payload is missing 'code' or 'keybinding'.")

    except json.JSONDecodeError:
        logging.error("Authenticated request failed during processing: Payload is not valid JSON.")
    except Exception as e:
        logging.error(f"An error occurred while processing the authenticated payload: {e}")