fix(resources): fix ResourceTypeSpec inheritance cycle detection for multi-level cycles #10633
@@ -0,0 +1,47 @@
|
||||
Feature: ResourceTypeSpec inheritance cycle detection for multi-level cycles
|
||||
ADR-042 defines single-inheritance for resource types.
|
||||
The validation should detect not only direct self-inheritance (A→A)
|
||||
but also multi-level cycles (A→B→A, A→B→C→A, etc.)
|
||||
|
||||
Background:
|
||||
Given a resource type registry for cycle detection
|
||||
|
||||
# ── Direct self-inheritance (existing behavior) ──────────────────────────
|
||||
|
||||
Scenario: Direct self-inheritance is rejected
|
||||
When I create a ResourceTypeSpec with name "acme/looper" inheriting "acme/looper"
|
||||
Then the creation should fail with "cannot inherit from itself"
|
||||
|
||||
# ── Two-level cycles (new behavior) ──────────────────────────────────────
|
||||
|
||||
Scenario: Two-level cycle A→B→A is detected
|
||||
Given a registered type "acme/alpha" inheriting from "acme/beta"
|
||||
When I create a ResourceTypeSpec with name "acme/beta" inheriting "acme/alpha"
|
||||
Then the creation should fail with either "circular" or "cycle"
|
||||
|
||||
# ── Three-level cycles (new behavior) ────────────────────────────────────
|
||||
|
||||
Scenario: Three-level cycle A→B→C→A is detected
|
||||
Given a registered type "acme/alpha" inheriting from "acme/beta"
|
||||
And a registered type "acme/beta" inheriting from "acme/gamma"
|
||||
When I create a ResourceTypeSpec with name "acme/gamma" inheriting "acme/alpha"
|
||||
Then the creation should fail with either "circular" or "cycle"
|
||||
|
||||
# ── Valid multi-level inheritance (should succeed) ──────────────────────
|
||||
|
||||
Scenario: Valid three-level chain A→B→C succeeds
|
||||
Given a registered type "acme/alpha" inheriting from "acme/beta"
|
||||
And a registered type "acme/beta" inheriting from "acme/gamma"
|
||||
When I create a ResourceTypeSpec with name "acme/gamma" inheriting nothing
|
||||
Then the creation should succeed
|
||||
|
||||
Scenario: Valid two-level chain A→B succeeds
|
||||
Given a registered type "acme/alpha" inheriting from "acme/beta"
|
||||
When I create a ResourceTypeSpec with name "acme/beta" inheriting nothing
|
||||
Then the creation should succeed
|
||||
|
||||
# ── Depth limit (DoS guard) ───────────────────────────────────────────────
|
||||
|
||||
Scenario: Inheritance chain exceeding the depth limit is rejected
|
||||
When I create a ResourceTypeSpec whose ancestry chain exceeds the depth limit
|
||||
Then the creation should fail with "exceeds"
|
||||
@@ -0,0 +1,133 @@
|
||||
"""Step definitions for ResourceTypeSpec inheritance cycle detection.
|
||||
|
||||
Tests that the validation detects multi-level inheritance cycles,
|
||||
not just direct self-inheritance.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
|
||||
from cleveragents.domain.models.core._resource_type_validation import (
|
||||
MAX_INHERITANCE_DEPTH,
|
||||
)
|
||||
from cleveragents.domain.models.core.resource_type import (
|
||||
ResourceTypeSpec,
|
||||
)
|
||||
|
||||
|
||||
@given("a resource type registry for cycle detection")
|
||||
def step_init_registry(context: Any) -> None:
|
||||
"""Initialize an empty registry for cycle detection tests."""
|
||||
context.type_registry = {} # type: ignore[attr-defined]
|
||||
context.creation_error = None # type: ignore[attr-defined]
|
||||
|
||||
|
||||
@given('a registered type "{name}" inheriting from "{parent}"')
|
||||
def step_register_type(context: Any, name: str, parent: str) -> None:
|
||||
"""Register a type in the test registry."""
|
||||
registry = context.type_registry # type: ignore[attr-defined]
|
||||
registry[name] = {
|
||||
"name": name,
|
||||
"resource_kind": "physical",
|
||||
"sandbox_strategy": "none",
|
||||
"inherits": parent,
|
||||
}
|
||||
|
||||
|
||||
@when('I create a ResourceTypeSpec with name "{name}" inheriting "{parent}"')
|
||||
def step_create_with_inherits(context: Any, name: str, parent: str) -> None:
|
||||
"""Attempt to create a ResourceTypeSpec with the given inheritance.
|
||||
|
||||
Passes the test registry as Pydantic validation context so that
|
||||
ResourceTypeSpec._validate_model() performs multi-level cycle detection
|
||||
through the production code path.
|
||||
"""
|
||||
registry: dict[str, Any] = getattr(context, "type_registry", {})
|
||||
ctx = {"type_registry": registry} if registry else None
|
||||
try:
|
||||
spec = ResourceTypeSpec.model_validate(
|
||||
{
|
||||
"name": name,
|
||||
"resource_kind": "physical",
|
||||
"sandbox_strategy": "none",
|
||||
|
|
||||
"inherits": parent,
|
||||
},
|
||||
context=ctx,
|
||||
)
|
||||
context.created_spec = spec # type: ignore[attr-defined]
|
||||
context.creation_error = None # type: ignore[attr-defined]
|
||||
except (ValueError, TypeError) as exc:
|
||||
context.creation_error = exc # type: ignore[attr-defined]
|
||||
|
||||
|
||||
@when('I create a ResourceTypeSpec with name "{name}" inheriting nothing')
|
||||
def step_create_without_inherits(context: Any, name: str) -> None:
|
||||
"""Attempt to create a ResourceTypeSpec without inheritance."""
|
||||
try:
|
||||
spec = ResourceTypeSpec.model_validate(
|
||||
{
|
||||
"name": name,
|
||||
"resource_kind": "physical",
|
||||
"sandbox_strategy": "none",
|
||||
"inherits": None,
|
||||
}
|
||||
)
|
||||
context.created_spec = spec # type: ignore[attr-defined]
|
||||
context.creation_error = None # type: ignore[attr-defined]
|
||||
except (ValueError, TypeError) as exc:
|
||||
context.creation_error = exc # type: ignore[attr-defined]
|
||||
|
||||
|
||||
@then('the creation should fail with "{fragment}"')
|
||||
def step_creation_failed(context: Any, fragment: str) -> None:
|
||||
"""Assert that creation failed with an error containing the fragment."""
|
||||
err = context.creation_error # type: ignore[attr-defined]
|
||||
assert err is not None, "Expected creation to fail but it succeeded"
|
||||
assert fragment.lower() in str(err).lower(), (
|
||||
f"Expected error to contain '{fragment}', got: {err}"
|
||||
)
|
||||
|
||||
|
||||
@then('the creation should fail with either "{fragment1}" or "{fragment2}"')
|
||||
def step_creation_failed_either(context: Any, fragment1: str, fragment2: str) -> None:
|
||||
"""Assert that creation failed with an error containing either fragment."""
|
||||
err = context.creation_error # type: ignore[attr-defined]
|
||||
assert err is not None, "Expected creation to fail but it succeeded"
|
||||
err_lower = str(err).lower()
|
||||
assert fragment1.lower() in err_lower or fragment2.lower() in err_lower, (
|
||||
f"Expected error to contain '{fragment1}' or '{fragment2}', got: {err}"
|
||||
)
|
||||
|
||||
|
||||
@then("the creation should succeed")
|
||||
def step_creation_succeeded(context: Any) -> None:
|
||||
"""Assert that creation succeeded."""
|
||||
err = context.creation_error # type: ignore[attr-defined]
|
||||
assert err is None, f"Expected creation to succeed but got error: {err}"
|
||||
assert hasattr(context, "created_spec"), "Expected created_spec to be set"
|
||||
|
||||
|
||||
@when("I create a ResourceTypeSpec whose ancestry chain exceeds the depth limit")
|
||||
def step_create_exceeds_depth(context: Any) -> None:
|
||||
"""Build a registry chain longer than MAX_INHERITANCE_DEPTH and attempt creation."""
|
||||
chain_len = MAX_INHERITANCE_DEPTH + 1
|
||||
registry: dict[str, Any] = {
|
||||
f"chain/t{i}": {"inherits": f"chain/t{i + 1}"} for i in range(chain_len)
|
||||
}
|
||||
try:
|
||||
spec = ResourceTypeSpec.model_validate(
|
||||
{
|
||||
"name": "chain/new",
|
||||
"resource_kind": "physical",
|
||||
"sandbox_strategy": "none",
|
||||
"inherits": "chain/t0",
|
||||
},
|
||||
context={"type_registry": registry},
|
||||
)
|
||||
context.created_spec = spec # type: ignore[attr-defined]
|
||||
context.creation_error = None # type: ignore[attr-defined]
|
||||
except (ValueError, TypeError) as exc:
|
||||
context.creation_error = exc # type: ignore[attr-defined]
|
||||
@@ -14,6 +14,7 @@ _BUILTIN_NAME_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9_-]*$")
|
||||
_NAMESPACED_RE = re.compile(r"^[a-zA-Z][a-zA-Z0-9_-]*/[a-zA-Z][a-zA-Z0-9_-]*$")
|
||||
|
||||
MAX_SCAN_DEPTH = 10
|
||||
MAX_INHERITANCE_DEPTH = 100
|
||||
|
||||
# All known built-in type names. Extracted here so that
|
||||
# ``ResourceTypeSpec.BUILTIN_NAMES`` stays a one-liner reference.
|
||||
@@ -272,3 +273,63 @@ def validate_self_referential(
|
||||
"and child_types) but has no scan_depth bound. "
|
||||
"Recursive types MUST specify scan_depth."
|
||||
)
|
||||
|
||||
|
||||
def detect_inheritance_cycles(
|
||||
name: str,
|
||||
inherits: str | None,
|
||||
registry: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Detect inheritance cycles in a resource type definition.
|
||||
|
||||
This function detects both direct self-inheritance (A→A) and
|
||||
multi-level cycles (A→B→A, A→B→C→A, etc.) by traversing the
|
||||
inheritance chain.
|
||||
|
||||
Args:
|
||||
name: The resource type name being validated.
|
||||
inherits: The parent type name (or None for root types).
|
||||
registry: Optional registry of existing types for cycle detection.
|
||||
If provided, multi-level cycles will be detected.
|
||||
If None, only direct self-inheritance is checked.
|
||||
|
||||
Raises:
|
||||
ValueError: If a cycle is detected.
|
||||
"""
|
||||
# Direct self-inheritance check (always performed)
|
||||
if inherits is not None and inherits == name:
|
||||
raise ValueError(f"'{name}' cannot inherit from itself.")
|
||||
|
||||
# Multi-level cycle detection (if registry is provided)
|
||||
if registry is not None and inherits is not None:
|
||||
visited: set[str] = {name}
|
||||
current = inherits
|
||||
depth = 0
|
||||
|
||||
|
HAL9001
commented
BLOCKING (security/reliability): The while-loop at this line traverses the inheritance chain without a depth limit. Add a MAX_INHERITANCE_DEPTH constant and raise ValueError if exceeded. BLOCKING (security/reliability): The while-loop at this line traverses the inheritance chain without a depth limit. Add a MAX_INHERITANCE_DEPTH constant and raise ValueError if exceeded.
|
||||
while current is not None:
|
||||
depth += 1
|
||||
if depth > MAX_INHERITANCE_DEPTH:
|
||||
raise ValueError(
|
||||
f"Inheritance chain for '{name}' exceeds the maximum "
|
||||
f"allowed depth of {MAX_INHERITANCE_DEPTH}."
|
||||
)
|
||||
if current in visited:
|
||||
# Cycle detected
|
||||
raise ValueError(
|
||||
f"Circular inheritance detected in '{name}': "
|
||||
f"inheritance chain would create a cycle through '{current}'."
|
||||
)
|
||||
|
||||
visited.add(current)
|
||||
|
||||
# Get the parent's parent
|
||||
entry = registry.get(current)
|
||||
if entry is None:
|
||||
# Parent not in registry, stop traversal
|
||||
break
|
||||
|
||||
# Extract inherits from the entry
|
||||
if isinstance(entry, dict):
|
||||
current = entry.get("inherits")
|
||||
else:
|
||||
current = getattr(entry, "inherits", None)
|
||||
|
||||
@@ -22,11 +22,21 @@ import re
|
||||
from enum import StrEnum
|
||||
from typing import Any, ClassVar
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
||||
from pydantic import (
|
||||
BaseModel,
|
||||
ConfigDict,
|
||||
Field,
|
||||
ValidationInfo,
|
||||
field_validator,
|
||||
model_validator,
|
||||
)
|
||||
|
||||
from cleveragents.domain.models.core._resource_type_validation import (
|
||||
BUILTIN_TYPE_NAMES,
|
||||
)
|
||||
from cleveragents.domain.models.core._resource_type_validation import (
|
||||
detect_inheritance_cycles as _detect_inheritance_cycles,
|
||||
)
|
||||
from cleveragents.domain.models.core._resource_type_validation import (
|
||||
validate_auto_discovery as _validate_auto_discovery,
|
||||
)
|
||||
@@ -308,7 +318,7 @@ class ResourceTypeSpec(BaseModel):
|
||||
# -- Cross-field validation -----------------------------------------------
|
||||
|
||||
@model_validator(mode="after")
|
||||
def _validate_model(self) -> ResourceTypeSpec:
|
||||
def _validate_model(self, info: ValidationInfo) -> ResourceTypeSpec:
|
||||
"""Cross-field validation for resource type constraints."""
|
||||
# Custom types must be namespaced
|
||||
if not self.built_in and "/" not in self.name:
|
||||
@@ -338,9 +348,9 @@ class ResourceTypeSpec(BaseModel):
|
||||
self.name, self.parent_types, self.child_types, self.auto_discovery
|
||||
)
|
||||
|
||||
# ADR-042 rule 3: no cycles (self-inheritance)
|
||||
if self.inherits is not None and self.inherits == self.name:
|
||||
raise ValueError(f"'{self.name}' cannot inherit from itself.")
|
||||
# ADR-042 rule 3: no cycles (self-inheritance and multi-level cycles)
|
||||
_registry = info.context.get("type_registry") if info.context else None
|
||||
_detect_inheritance_cycles(self.name, self.inherits, _registry)
|
||||
# ADR-042 rule 4: built-in must not inherit from custom
|
||||
if self.built_in and self.inherits is not None and "/" in self.inherits:
|
||||
raise ValueError(
|
||||
|
||||
Reference in New Issue
Block a user
BLOCKING: The step at this line calls detect_inheritance_cycles() with a registry BEFORE creating ResourceTypeSpec. The test validates the standalone function, not _validate_model() integration.