lablink_allocator_service.validate_config

lablink_allocator_service.validate_config

Config validation CLI for LabLink Allocator Service.

This module provides a command-line tool to validate configuration files against the Hydra/OmegaConf schema before deployment. This enables fail-fast validation during CI/CD pipelines.

Attributes

MIN_ADMIN_PASSWORD_LENGTH = 12 module-attribute

PUBLIC_HOSTNAME_HINT = "must be a bare hostname such as 'lab.smithlab.org' — no scheme (drop the https://), port, path, or whitespace" module-attribute

PUBLIC_HOSTNAME_RE = re.compile('^(?=.{1,253}$)[A-Za-z0-9]([A-Za-z0-9-]{0,61}[A-Za-z0-9])?(\\.[A-Za-z0-9]([A-Za-z0-9-]{0,61}[A-Za-z0-9])?)+$') module-attribute

VALID_CONNECTIVITY = ('lan_direct', 'mesh_overlay', 'reverse_tunnel') module-attribute

VALID_PARTICIPANT_EXPOSURE = ('none', 'tailscale_funnel', 'cloudflare_tunnel') module-attribute

VALID_PROVIDERS = ('aws', 'manual') module-attribute

WEAK_ADMIN_PASSWORDS = frozenset({'123456', 'admin', 'password', 'changeme', 'placeholder_admin_password', ''}) module-attribute

logger = logging.getLogger(__name__) module-attribute

Functions:

get_config_errors(cfg)

Return a list of DNS/SSL validation errors (empty = valid).

This is the single source of truth for DNS/SSL validation rules, shared by both the allocator's validate_config CLI and the lablink CLI's config validation.

Parameters:

Name	Type	Description	Default
cfg		Any config object with .dns and .ssl attributes (works with both Config dataclass and DictConfig).	required

Returns:

Type	Description
list	List of error message strings.

Source code in packages/allocator/src/lablink_allocator_service/validate_config.py

def get_config_errors(cfg) -> list:
    """Return a list of DNS/SSL validation errors (empty = valid).

    This is the single source of truth for DNS/SSL validation rules,
    shared by both the allocator's validate_config CLI and the
    lablink CLI's config validation.

    Args:
        cfg: Any config object with .dns and .ssl attributes
            (works with both Config dataclass and DictConfig).

    Returns:
        List of error message strings.
    """
    errors = []

    # Top-level provider must be one of the registered providers.
    # `getattr` keeps the validator robust if a legacy DictConfig without
    # the `provider` field is passed in (defaults to the registry default).
    provider = getattr(cfg, "provider", "aws")
    if provider not in VALID_PROVIDERS:
        errors.append(
            f"provider must be one of: {', '.join(VALID_PROVIDERS)} (got '{provider}')"
        )

    # manual.connectivity must be a known value. Either mesh_overlay
    # (to resolve overlay hostnames — see
    # MeshOverlayClientConnectivity._resolve_overlay_host) or
    # participant_exposure == "tailscale_funnel" (to publish the
    # allocator's own hostname) requires a tailnet domain.
    manual_cfg = getattr(cfg, "manual", None)
    if manual_cfg is not None:
        connectivity = getattr(manual_cfg, "connectivity", "lan_direct")
        if connectivity not in VALID_CONNECTIVITY:
            errors.append(
                f"manual.connectivity must be one of: "
                f"{', '.join(VALID_CONNECTIVITY)} (got '{connectivity}')"
            )

        participant_exposure = getattr(manual_cfg, "participant_exposure", "none")
        if participant_exposure not in VALID_PARTICIPANT_EXPOSURE:
            errors.append(
                f"manual.participant_exposure must be one of: "
                f"{', '.join(VALID_PARTICIPANT_EXPOSURE)} "
                f"(got '{participant_exposure}')"
            )

        # lan_direct sends the participant's browser straight to a client's
        # LAN IP (ws://<client-ip>:6080 — see LANDirectClientConnectivity),
        # bypassing the allocator entirely. That's unreachable off-LAN and,
        # once the allocator itself is publicly exposed, actively blocked as
        # mixed content by the browser (ws:// from an https:// page).
        # mesh_overlay doesn't have this problem — it proxies sessions
        # through the allocator's own nginx, which any exposure mode
        # publishes. Keyed on "any exposure" rather than on Funnel
        # specifically: the reasoning is about being publicly reachable,
        # not about which tunnel does the publishing.
        if participant_exposure != "none" and connectivity == "lan_direct":
            errors.append(
                f"manual.participant_exposure is '{participant_exposure}' but "
                "manual.connectivity is 'lan_direct' — participant sessions "
                "would connect directly to a client's LAN IP, which is "
                "unreachable off-LAN and blocked as mixed content from the "
                "HTTPS page. Use manual.connectivity: mesh_overlay "
                "instead, which proxies sessions through the allocator."
            )

        # cloudflare_tunnel publishes at an admin-chosen hostname, so that
        # hostname is not optional. The tunnel token is checked at deploy
        # time instead (it lives in .env, never in config.yaml).
        if participant_exposure == "cloudflare_tunnel":
            public_hostname = getattr(manual_cfg, "public_hostname", "")
            if not public_hostname:
                errors.append(
                    "manual.participant_exposure is 'cloudflare_tunnel' but "
                    "manual.public_hostname is empty — set it to the hostname "
                    "configured as the tunnel's public hostname in Cloudflare "
                    "(e.g. lab.smithlab.org)."
                )
            elif not is_valid_public_hostname(public_hostname):
                errors.append(
                    f"manual.public_hostname {PUBLIC_HOSTNAME_HINT} "
                    f"(got '{public_hostname}')"
                )

        needs_tailnet = (
            connectivity == "mesh_overlay" or participant_exposure == "tailscale_funnel"
        )
        if needs_tailnet and not getattr(manual_cfg, "overlay_tailnet", ""):
            errors.append(
                "manual.overlay_tailnet is required when manual.connectivity "
                "is 'mesh_overlay' or manual.participant_exposure is "
                "'tailscale_funnel' (e.g. 'example.ts.net')"
            )

        if participant_exposure == "tailscale_funnel":
            admin_password = getattr(getattr(cfg, "app", None), "admin_password", "")
            # MISSING_SECRET is AppConfig.admin_password's dataclass default —
            # the sentinel for "not yet resolved" (the wizard never collects
            # it; resolve_admin_credentials fills it in at deploy time, and
            # THAT resolved value is what deploy_compose.py's own preflight
            # gate checks). Treating the sentinel as "weak" would block
            # `lablink configure`'s ReviewScreen on every fresh config,
            # before the operator has had any chance to set a real password.
            if admin_password != MISSING_SECRET and is_weak_admin_password(
                admin_password
            ):
                errors.append(
                    "manual.participant_exposure is 'tailscale_funnel' but "
                    "app.admin_password is empty, a known example value, or "
                    "shorter than 12 characters — a Funnel-exposed allocator "
                    "is scanned by bots within minutes; set a strong "
                    "admin_password"
                )

    # DNS enabled requires non-empty domain
    if cfg.dns.enabled and not cfg.dns.domain:
        errors.append("DNS enabled requires non-empty domain field")

    # Validate domain format
    is_valid, error_msg = validate_domain_format(cfg.dns.domain)
    if not is_valid:
        errors.append(error_msg)

    # SSL (non-"none") requires DNS
    if cfg.ssl.provider != "none" and not cfg.dns.enabled:
        errors.append(
            'SSL requires DNS to be enabled (use provider="none" for HTTP-only)'
        )

    # Let's Encrypt requires email
    if cfg.ssl.provider == "letsencrypt" and not cfg.ssl.email:
        errors.append("Let's Encrypt requires email address")

    # ACM requires certificate_arn
    if cfg.ssl.provider == "acm" and not cfg.ssl.certificate_arn:
        errors.append("ACM provider requires certificate_arn")

    # CloudFlare SSL requires external DNS (terraform_managed=false)
    if cfg.ssl.provider == "cloudflare" and cfg.dns.terraform_managed:
        errors.append(
            "CloudFlare SSL requires terraform_managed=false (external DNS management)"
        )

    return errors

is_valid_public_hostname(hostname)

True if hostname is a bare FQDN safe to interpolate into a URL.

Empty is not accepted here; callers report that separately, because "you left it blank" and "you typed a URL" need different guidance.

Source code in packages/allocator/src/lablink_allocator_service/validate_config.py

def is_valid_public_hostname(hostname: str) -> bool:
    """True if *hostname* is a bare FQDN safe to interpolate into a URL.

    Empty is *not* accepted here; callers report that separately, because
    "you left it blank" and "you typed a URL" need different guidance.
    """
    return bool(PUBLIC_HOSTNAME_RE.match(hostname))

is_weak_admin_password(password)

True if password is empty, a known example/default value, or shorter than the minimum length required once the allocator is reachable from the public internet.

Source code in packages/allocator/src/lablink_allocator_service/validate_config.py

def is_weak_admin_password(password: str) -> bool:
    """True if *password* is empty, a known example/default value, or
    shorter than the minimum length required once the allocator is
    reachable from the public internet."""
    if not password:
        return True
    if password.lower() in WEAK_ADMIN_PASSWORDS:
        return True
    return len(password) < MIN_ADMIN_PASSWORD_LENGTH

main()

Main entry point for the config validation CLI.

Source code in packages/allocator/src/lablink_allocator_service/validate_config.py

def main():
    """Main entry point for the config validation CLI."""
    logging.basicConfig(
        level=logging.WARNING,
        format="%(message)s",
    )

    parser = argparse.ArgumentParser(
        description="Validate LabLink allocator configuration file against schema",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  # Validate deployment config
  lablink-validate-config config/config.yaml

  # Validate runtime config in Docker
  lablink-validate-config /config/config.yaml

  # Validate bundled config
  lablink-validate-config \\
      packages/allocator/src/lablink_allocator_service/conf/config.yaml

NOTE: Config file MUST be named 'config.yaml' for schema validation.

Exit codes:
  0 - Config is valid
  1 - Config is invalid or error occurred
        """,
    )

    parser.add_argument(
        "config_path",
        nargs="?",
        default="/config/config.yaml",
        help="Path to the config.yaml file to validate (default: /config/config.yaml)",
    )

    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Enable verbose output",
    )

    args = parser.parse_args()

    # Set logging level based on verbosity
    if args.verbose:
        logging.getLogger().setLevel(logging.INFO)
        logging.getLogger("lablink_allocator_service").setLevel(logging.INFO)

    # Validate the configuration
    is_valid, message = validate_config(args.config_path)

    # Print the result
    print(message)

    # Exit with appropriate code
    sys.exit(0 if is_valid else 1)

validate_config(config_path)

Validate a configuration file against the schema.

Parameters:

Name	Type	Description	Default
config_path	str	Path to the config.yaml file to validate	required

Returns:

Type	Description
Tuple[bool, str]	Tuple of (is_valid, message): - is_valid: True if config is valid, False otherwise - message: Success or error message

Source code in packages/allocator/src/lablink_allocator_service/validate_config.py

def validate_config(config_path: str) -> Tuple[bool, str]:
    """Validate a configuration file against the schema.

    Args:
        config_path: Path to the config.yaml file to validate

    Returns:
        Tuple of (is_valid, message):
            - is_valid: True if config is valid, False otherwise
            - message: Success or error message
    """
    path = Path(config_path)

    # Check if file exists
    if not path.exists():
        return False, f"[FAIL] Config file not found: {config_path}"

    if not path.is_file():
        return False, f"[FAIL] Config path is not a file: {config_path}"

    # Require config.yaml filename for Hydra schema matching
    if path.name != "config.yaml":
        return False, (
            f"[FAIL] Config file must be named 'config.yaml'\n"
            f"       Found: {path.name}\n"
            f"       Rename your file to enable strict schema validation"
        )

    try:
        # Use get_config() with explicit path - it validates automatically
        cfg = get_config(config_path=path.as_posix())

        # Run logic validation
        is_valid, error_msg = validate_config_logic(cfg)
        if not is_valid:
            return False, error_msg

        return True, "[PASS] Config validation passed"

    except ConfigCompositionException as e:
        # This is the error from your Docker logs - extract the key info
        error_msg = (
            "[FAIL] Config validation failed: Error merging config with schema\n"
        )
        error_str = str(e)

        # Try to extract the key that caused the problem
        if "Key '" in error_str and "' not in" in error_str:
            # Extract key name from error message
            key_start = error_str.find("Key '") + 5
            key_end = error_str.find("'", key_start)
            bad_key = error_str[key_start:key_end]
            error_msg += f"       Unknown key: '{bad_key}'\n"
            error_msg += "       This key is not defined in the Config schema\n"
        else:
            error_msg += f"       {error_str}\n"

        return False, error_msg

    except ConfigKeyError as e:
        error_msg = "[FAIL] Config validation failed: Unknown configuration key\n"
        error_msg += f"       Key '{e.key}' not found in schema"
        if hasattr(e, "full_key") and e.full_key:
            error_msg += f"\n       Full key path: {e.full_key}"
        if hasattr(e, "object_type") and e.object_type:
            type_name = (
                e.object_type.__name__
                if hasattr(e.object_type, "__name__")
                else str(e.object_type)
            )
            error_msg += f"\n       Expected in schema: {type_name}"
        error_msg += "\n"
        return False, error_msg

    except ValidationError as e:
        error_msg = "[FAIL] Config validation failed: Schema validation error\n"
        error_msg += f"       {str(e)}\n"
        return False, error_msg

    except Exception as e:
        logger.exception("Unexpected error during config validation")
        error_msg = f"[FAIL] Config validation failed: {type(e).__name__}\n"
        error_msg += f"       {str(e)}\n"
        return False, error_msg

validate_config_logic(cfg)

Validate configuration logic and dependencies.

Parameters:

Name	Type	Description	Default
cfg	DictConfig	Loaded Hydra/OmegaConf configuration object.	required

Returns:

Type	Description
Tuple[bool, str]	Tuple of (is_valid, error_message)

Source code in packages/allocator/src/lablink_allocator_service/validate_config.py

def validate_config_logic(cfg: DictConfig) -> Tuple[bool, str]:
    """Validate configuration logic and dependencies.

    Args:
        cfg: Loaded Hydra/OmegaConf configuration object.

    Returns:
        Tuple of (is_valid, error_message)
    """
    errors = get_config_errors(cfg)

    if errors:
        error_msg = "[FAIL] Config validation failed:\n"
        for error in errors:
            error_msg += f"       - {error}\n"
        return False, error_msg

    return True, ""

validate_domain_format(domain)

Validate domain format to prevent malformed domains.

Parameters:

Name	Type	Description	Default
domain	str	Domain name to validate	required

Returns:

Type	Description
Tuple[bool, str]	Tuple of (is_valid, error_message)

Source code in packages/allocator/src/lablink_allocator_service/validate_config.py

def validate_domain_format(domain: str) -> Tuple[bool, str]:
    """Validate domain format to prevent malformed domains.

    Args:
        domain: Domain name to validate

    Returns:
        Tuple of (is_valid, error_message)
    """
    if not domain:
        return True, ""  # Empty is allowed when DNS disabled

    # Check for leading/trailing dots
    if domain.startswith("."):
        return False, "Domain cannot start with a dot"
    if domain.endswith("."):
        return False, "Domain cannot end with a dot"

    return True, ""