changedetection.io/changedetectionio/api/__init__.py

import functools
from flask import request, abort
from loguru import logger

@functools.cache
def build_merged_spec_dict():
    """
    Load the base OpenAPI spec and merge in any per-processor api.yaml extensions.

    Each processor can provide an api.yaml file alongside its __init__.py that defines
    additional schemas (e.g., processor_config_restock_diff). These are merged into
    WatchBase.properties so the spec accurately reflects what the API accepts.

    Plugin processors (via pluggy) are also supported - they just need an api.yaml
    next to their processor module.

    Returns the merged dict (cached - do not mutate the returned value).
    """
    import os
    import yaml

    spec_path = os.path.join(os.path.dirname(__file__), '../../docs/api-spec.yaml')
    if not os.path.exists(spec_path):
        spec_path = os.path.join(os.path.dirname(__file__), '../docs/api-spec.yaml')

    with open(spec_path, 'r', encoding='utf-8') as f:
        spec_dict = yaml.safe_load(f)

    try:
        from changedetectionio.processors import find_processors, get_parent_module
        for module, proc_name in find_processors():
            parent = get_parent_module(module)
            if not parent or not hasattr(parent, '__file__'):
                continue
            api_yaml_path = os.path.join(os.path.dirname(parent.__file__), 'api.yaml')
            if not os.path.exists(api_yaml_path):
                continue
            with open(api_yaml_path, 'r', encoding='utf-8') as f:
                proc_spec = yaml.safe_load(f)
            # Merge schemas
            proc_schemas = proc_spec.get('components', {}).get('schemas', {})
            spec_dict['components']['schemas'].update(proc_schemas)
            # Inject processor_config_{name} into WatchBase if the schema is defined
            schema_key = f'processor_config_{proc_name}'
            if schema_key in proc_schemas:
                spec_dict['components']['schemas']['WatchBase']['properties'][schema_key] = {
                    '$ref': f'#/components/schemas/{schema_key}'
                }
            # Append x-code-samples from processor paths into existing path operations
            for path, path_item in proc_spec.get('paths', {}).items():
                if path not in spec_dict.get('paths', {}):
                    continue
                for method, operation in path_item.items():
                    if method not in spec_dict['paths'][path]:
                        continue
                    if 'x-code-samples' in operation:
                        existing = spec_dict['paths'][path][method].get('x-code-samples', [])
                        spec_dict['paths'][path][method]['x-code-samples'] = existing + operation['x-code-samples']
    except Exception as e:
        logger.warning(f"Failed to merge processor API specs: {e}")

    return spec_dict


@functools.cache
def get_openapi_spec():
    """Lazy load OpenAPI spec and dependencies only when validation is needed."""
    from openapi_core import OpenAPI  # Lazy import - saves ~10.7 MB on startup
    return OpenAPI.from_dict(build_merged_spec_dict())

@functools.cache
def get_openapi_schema_dict():
    """
    Get the raw OpenAPI spec dictionary for schema access.

    Used by Import endpoint to validate and convert query parameters.
    Returns the merged YAML dict (not the OpenAPI object).
    """
    return build_merged_spec_dict()

@functools.cache
def _resolve_schema_properties(schema_name):
    """
    Generic helper to resolve schema properties, including allOf inheritance.

    Args:
        schema_name: Name of the schema (e.g., 'WatchBase', 'Watch', 'Tag')

    Returns:
        dict: All properties including inherited ones from $ref schemas
    """
    spec_dict = get_openapi_schema_dict()
    schema = spec_dict['components']['schemas'].get(schema_name, {})

    properties = {}

    # Handle allOf (schema inheritance)
    if 'allOf' in schema:
        for item in schema['allOf']:
            # Resolve $ref to parent schema
            if '$ref' in item:
                ref_path = item['$ref'].split('/')[-1]
                ref_schema = spec_dict['components']['schemas'].get(ref_path, {})
                properties.update(ref_schema.get('properties', {}))
            # Add schema-specific properties
            if 'properties' in item:
                properties.update(item['properties'])
    else:
        # Direct properties (no inheritance)
        properties = schema.get('properties', {})

    return properties


@functools.cache
def get_watch_schema_properties():
    """
    Extract watch schema properties from OpenAPI spec for Import endpoint.

    Returns WatchBase properties (all writable Watch fields).
    """
    return _resolve_schema_properties('WatchBase')

# Import readonly field utilities from shared module (avoids circular dependencies with model layer)
from changedetectionio.model.schema_utils import get_readonly_watch_fields, get_readonly_tag_fields

@functools.cache
def get_tag_schema_properties():
    """
    Extract Tag schema properties from OpenAPI spec.

    Returns WatchBase properties + Tag-specific properties (overrides_watch).
    """
    return _resolve_schema_properties('Tag')

def validate_openapi_request(operation_id):
    """Decorator to validate incoming requests against OpenAPI spec."""
    def decorator(f):
        @functools.wraps(f)
        def wrapper(*args, **kwargs):
            from werkzeug.exceptions import BadRequest
            try:
                # Skip OpenAPI validation for GET requests since they don't have request bodies
                if request.method.upper() != 'GET':
                    # Lazy import - only loaded when actually validating a request
                    from openapi_core.contrib.flask import FlaskOpenAPIRequest
                    from openapi_core.templating.paths.exceptions import ServerNotFound, PathNotFound, PathError

                    spec = get_openapi_spec()
                    openapi_request = FlaskOpenAPIRequest(request)
                    result = spec.unmarshal_request(openapi_request)
                    if result.errors:
                        error_details = []
                        for error in result.errors:
                            # Skip path/server validation errors for reverse proxy compatibility
                            # Flask routing already validates that endpoints exist (returns 404 if not).
                            # OpenAPI validation here is primarily for request body schema validation.
                            # When behind nginx/reverse proxy, URLs may have path prefixes that don't
                            # match the OpenAPI server definitions, causing false positives.
                            if isinstance(error, PathError):
                                logger.debug(f"API Call - Skipping path/server validation (delegated to Flask): {error}")
                                continue

                            error_str = str(error)
                            # Extract detailed schema errors from __cause__
                            if hasattr(error, '__cause__') and hasattr(error.__cause__, 'schema_errors'):
                                for schema_error in error.__cause__.schema_errors:
                                    field = '.'.join(str(p) for p in schema_error.path) if schema_error.path else 'body'
                                    msg = schema_error.message if hasattr(schema_error, 'message') else str(schema_error)
                                    error_details.append(f"{field}: {msg}")
                            else:
                                error_details.append(error_str)

                        # Only raise if we have actual validation errors (not path/server issues)
                        if error_details:
                            logger.error(f"API Call - Validation failed: {'; '.join(error_details)}")
                            raise BadRequest(f"Validation failed: {'; '.join(error_details)}")
            except BadRequest:
                # Re-raise BadRequest exceptions (validation failures)
                raise
            except Exception as e:
                # If OpenAPI spec loading fails, log but don't break existing functionality
                logger.critical(f"OpenAPI validation warning for {operation_id}: {e}")
                abort(500)
            return f(*args, **kwargs)
        return wrapper
    return decorator

# Import all API resources
from .Watch import Watch, WatchHistory, WatchSingleHistory, WatchHistoryDiff, CreateWatch, WatchFavicon
from .Tags import Tags, Tag
from .Import import Import
from .SystemInfo import SystemInfo
from .Spec import Spec
from .Notifications import Notifications