fix(resources): fix ResourceTypeSpec inheritance cycle detection for multi-level cycles #10633

Merged
HAL9000 merged 5 commits from fix/v360/resource-type-cycle-detection into master 2026-06-06 03:51:29 +00:00
4 changed files with 256 additions and 5 deletions
@@ -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 (AA)
but also multi-level cycles (ABA, ABCA, 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",
Outdated
Review

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.

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.
"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
Outdated
Review

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:
1
@@ -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(