integrations/mock_server/src/route_configuration.py

# Copyright (c) 2025 Project CHIP Authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import json
import logging
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Dict, List, Optional

log = logging.getLogger(__name__)


@dataclass
class RouteResponse:
    status: int
    headers: Dict[str, str]
    body: Any  # Dict for inline JSON, or bytes for $ref content (raw file bytes)


@dataclass
class QueryConfig:
    params: Dict[str, Any]


@dataclass
class Route:
    method: str
    path: str
    response: RouteResponse
    body: Optional[Any] = None
    query: Optional[QueryConfig] = None


@dataclass
class Configuration:
    routing: List[Route] = field(default_factory=list)


def load_configurations(config_path: Path, routing_config_dir: Path) -> Configuration:
    """
    Load and combine configuration files from specified paths.

    Args:
        config_path (Path): Path to the main configuration file.
        routing_config_dir (Path): Directory containing routing configuration files.

    Returns:
        Configuration: A Configuration object containing all merged routing configurations.

    Raises:
        ValueError: If config_path is not a file or routing_config_dir is not a directory.
    """
    if not config_path.is_file():
        raise ValueError(f"'{config_path}' is not a file")

    if not routing_config_dir.is_dir():
        raise ValueError(f"'{routing_config_dir}' is not a directory")

    # Load routing configuration
    routing_config: List[Route] = load_routing_configuration_dir(routing_config_dir)

    # Create and return the Configuration object
    return Configuration(routing=routing_config)


def load_routing_configuration_dir(directory: Path) -> List[Route]:
    """
    Load and merge all routing configuration files from a specified directory.

    Args:
        directory (Path): Directory path containing JSON configuration files.

    Returns:
        List[Route]: A list of Route objects containing all merged routing configurations.

    Raises:
        ValueError: If the specified directory path is not a directory.

    Note:
        - Processes all .json files in the specified directory
        - Continues processing even if one file fails to load, logging the error
    """
    if not directory.is_dir():
        raise ValueError(f"'{directory}' is not a directory")

    all_routes: List[Route] = []

    # Process all JSON files in the directory
    for file_path in directory.glob("*.json"):
        try:
            routes = load_routing_configuration_file(file_path)
            all_routes.extend(routes)
        except Exception as e:
            print(f"Error loading configuration from {file_path}: {str(e)}")
            continue

    return all_routes


def resolve_ref(body: Dict[str, Any], base_path: Path) -> bytes:
    """
    Resolve $ref reference and return raw file content.

    Args:
        body (Dict[str, Any]): The body object containing a $ref key.
        base_path (Path): The directory path to resolve relative references from.

    Returns:
        bytes: The raw file content as bytes.

    Example:
        body = {"$ref": "../static/tc-65521-32769-v1.json"}
        content = resolve_ref(body, Path("configurations/fake_product_server"))
        # Returns raw contents of configurations/static/tc-65521-32769-v1.json
    """
    ref_path_str = body["$ref"]
    ref_path = (base_path / ref_path_str).resolve()

    # To prevent path traversal, ensure the resolved path is within the parent directory of the config file.
    config_root = base_path.parent.resolve()
    try:
        ref_path.relative_to(config_root)
    except ValueError:
        log.error("Path traversal attempt in $ref: '%s' resolves to a path outside of '%s'", ref_path_str, config_root)
        raise ValueError("Invalid $ref path: path traversal is not allowed.")

    log.debug("Resolving $ref: %s -> %s", ref_path_str, ref_path)

    try:
        with open(ref_path, "rb") as ref_file:
            return ref_file.read()
    except FileNotFoundError:
        log.error("Referenced file not found: %s", ref_path)
        raise


def load_routing_configuration_file(file_path: Path) -> List[Route]:
    """
    Load and parse a single routing configuration JSON file.

    Args:
        file_path (Path): Path to the JSON configuration file.

    Returns:
        List[Route]: A list of Route objects parsed from the configuration file.

    Raises:
        FileNotFoundError: If the configuration file doesn't exist.
        json.JSONDecodeError: If the file contains invalid JSON.
        KeyError: If the required 'routes' key is missing in the configuration.

    Format:
        The JSON file should contain:
        {
            "routes": [
                {
                    "method": str,
                    "path": str,
                    "response": {
                        "status": int,
                        "headers": dict[str, str],
                        "body": dict[str, Any] | {"$ref": "path/to/file.json"}
                    },
                    "body": Any,  # Optional
                    "query": dict[str, Any]  # Optional
                }
            ]
        }

    Note:
        The "body" field in "response" can contain a "$ref" key pointing to an
        external JSON file. The reference is resolved relative to the config file's
        directory. This allows response bodies to be stored as separate static files
        that can be hosted independently (e.g., on GitHub for DCL references).
    """
    try:
        with open(file_path, "r") as file:
            config = json.load(file)
        if "routes" not in config:
            log.error("Missing required 'routes' key in configuration file: %s", file_path)
            raise KeyError("Configuration file must contain 'routes' key")

        log.debug("Routes configuration loaded successfully from %s", file_path)
        routes = []
        base_path = file_path.parent

        for route_config in config["routes"]:
            # Get the body, resolving $ref if present
            body = route_config["response"].get("body", {})
            if isinstance(body, dict) and "$ref" in body:
                # $ref: load raw file content as bytes (preserves exact bytes for hash verification)
                resolved_body = resolve_ref(body, base_path)
            else:
                # Inline body: keep as-is (dict or other)
                resolved_body = body

            # Create RouteResponse object
            response = RouteResponse(
                status=route_config["response"]["status"],
                headers=route_config["response"].get("headers", {}),
                body=resolved_body,
            )

            # Create QueryConfig if query params exist
            query = None
            if "query" in route_config:
                query = QueryConfig(params=route_config["query"])

            # Create Route object
            route = Route(
                method=route_config["method"],
                path=route_config["path"],
                response=response,
                body=route_config.get("body"),
                query=query,
            )
            routes.append(route)

        return routes

    except FileNotFoundError:
        log.error("Configuration file not found: %s", file_path)
        raise
    except json.JSONDecodeError as e:
        log.error("Invalid JSON in configuration file: %s", file_path)
        raise e