Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion backend/infrahub/api/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -156,7 +156,7 @@ def evaluate_candidate_schemas(
try:
schema = _merge_candidate_schemas(schemas=schemas_to_evaluate.schemas)

candidate_schema.load_schema(schema=schema)
candidate_schema.load_schema(schema=schema, loading_from_api=True)
candidate_schema.process()

schema_diff = branch_schema.diff(other=candidate_schema)
Expand Down
7 changes: 7 additions & 0 deletions backend/infrahub/core/constants/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,13 @@
"hfid",
]

RESERVED_ATTR_REL_HIERARCHICAL_NAMES = [
"parent",
"children",
"ancestors",
"descendants",
]

RESERVED_ATTR_GEN_NAMES = ["type"]

NULL_VALUE = "NULL"
Expand Down
11 changes: 10 additions & 1 deletion backend/infrahub/core/schema/relationship_schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from enum import Enum
from typing import TYPE_CHECKING, Any

from pydantic import BaseModel
from pydantic import BaseModel, PrivateAttr

from infrahub import config
from infrahub.core.constants import RelationshipDirection
Expand All @@ -23,6 +23,15 @@
class RelationshipSchema(GeneratedRelationshipSchema):
_exclude_from_hash: list[str] = ["filters"]
_sort_by: list[str] = ["name"]
_loaded_from_api: bool = PrivateAttr(default=False)

@property
def loaded_from_api(self) -> bool:
return self._loaded_from_api

@loaded_from_api.setter
def loaded_from_api(self, value: bool) -> None:
self._loaded_from_api = value
Comment on lines +26 to +34
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this violates the Single Responsibility Principle, which is something we do all over the place, but we are trying not to add any more
the RelationshipSchema should not need data indicating whether it was loaded from the API or not
if we need to do validation of data from the API before it is instantiated into a RelationshipSchema, then that should happen in the API endpoint somewhere


@property
def is_attribute(self) -> bool:
Expand Down
30 changes: 29 additions & 1 deletion backend/infrahub/core/schema/schema_branch.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
OBJECT_TEMPLATE_NAME_ATTR,
OBJECT_TEMPLATE_RELATIONSHIP_NAME,
RESERVED_ATTR_GEN_NAMES,
RESERVED_ATTR_REL_HIERARCHICAL_NAMES,
RESERVED_ATTR_REL_NAMES,
RESTRICTED_NAMESPACES,
BranchSupportType,
Expand Down Expand Up @@ -465,7 +466,7 @@ def generate_fields_for_display_label(self, name: str) -> dict | None:

return fields or None

def load_schema(self, schema: SchemaRoot) -> None:
def load_schema(self, schema: SchemaRoot, loading_from_api: bool = False) -> None:
"""Load a SchemaRoot object and store all NodeSchema or GenericSchema.

In the current implementation, if a schema object present in the SchemaRoot already exist, it will be overwritten.
Expand All @@ -491,6 +492,10 @@ def load_schema(self, schema: SchemaRoot) -> None:
except SchemaNotFoundError:
self.set(name=item.kind, schema=item)

if loading_from_api:
for relationship in item.relationships:
relationship.loaded_from_api = True

Comment on lines +495 to +498
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🔴 Critical

Bug: loaded_from_api set on the wrong object; extensions not flagged

Currently, you set relationship.loaded_from_api on item (incoming), but for updates you store new_item in the cache, so the flag isn’t present on the stored relationships. Also, relationships introduced via schema.extensions.nodes are never flagged, allowing a bypass.

Fix by marking the relationships on the stored object, limited to those provided by the incoming payload; and handle extensions similarly.

         for item in schema.nodes + schema.generics:
             try:
                 if item.id:
                     new_item = self.get_by_id(id=item.id)
                     if new_item.kind != item.kind:
                         self.delete(name=new_item.kind)
                 else:
                     new_item = self.get(name=item.kind)

                 if (new_item.is_node_schema and not item.is_node_schema) or (
                     new_item.is_generic_schema and not item.is_generic_schema
                 ):
                     current_node_type = "Node" if new_item.is_node_schema else "Generic"
                     raise ValidationError(
                         f"{item.kind} already exist in the schema as a {current_node_type}. Either rename it or delete the existing one."
                     )
-                new_item.update(item)
-                self.set(name=item.kind, schema=new_item)
+                new_item.update(item)
+                if loading_from_api:
+                    incoming_rel_names = {r.name for r in item.relationships}
+                    for rel in new_item.relationships:
+                        if rel.name in incoming_rel_names:
+                            rel.loaded_from_api = True
+                self.set(name=item.kind, schema=new_item)
             except SchemaNotFoundError:
-                self.set(name=item.kind, schema=item)
-
-            if loading_from_api:
-                for relationship in item.relationships:
-                    relationship.loaded_from_api = True
+                if loading_from_api:
+                    for relationship in item.relationships:
+                        relationship.loaded_from_api = True
+                self.set(name=item.kind, schema=item)

         for node_extension in schema.extensions.nodes:
             new_item = self.get(name=node_extension.kind)
-            new_item.update(node_extension)
-            self.set(name=node_extension.kind, schema=new_item)
+            new_item.update(node_extension)
+            if loading_from_api and node_extension.relationships:
+                ext_rel_names = {r.name for r in node_extension.relationships}
+                for rel in new_item.relationships:
+                    if rel.name in ext_rel_names:
+                        rel.loaded_from_api = True
+            self.set(name=node_extension.kind, schema=new_item)

Also applies to: 499-503

🤖 Prompt for AI Agents
In backend/infrahub/core/schema/schema_branch.py around lines 495-498 (and
similarly 499-503), the code currently sets loaded_from_api on the incoming
item.relationships which is never the object stored in the cache; change it to
set loaded_from_api on the relationships of the stored object (new_item)
instead, and only for relationships that were present in the incoming payload
(match by relationship name or id from the payload). Additionally, detect any
relationships added via schema.extensions.nodes and mark those corresponding
relationships on the stored object as loaded_from_api as well so extensions
cannot bypass the flagging.

for node_extension in schema.extensions.nodes:
new_item = self.get(name=node_extension.kind)
new_item.update(node_extension)
Expand Down Expand Up @@ -518,6 +523,7 @@ def process_pre_validation(self) -> None:
self.add_hierarchy_node()

def process_validate(self) -> None:
self.validate_hierarchical_nodes_restricted_words()
self.validate_names()
self.validate_python_keywords()
self.validate_kinds()
Expand Down Expand Up @@ -989,6 +995,28 @@ def validate_names(self) -> None:
):
raise ValueError(f"{node.kind}: {rel.name} isn't allowed as a relationship name.")

def validate_hierarchical_nodes_restricted_words(self) -> None:
for name in self.all_names:
node = self.get(name=name, duplicate=False)
is_hierarchical_node = (isinstance(node, GenericSchema) and node.hierarchical) or (
isinstance(node, NodeSchema) and node.hierarchy
)

if not is_hierarchical_node or node.kind in INTERNAL_SCHEMA_NODE_KINDS:
continue
Comment on lines +1001 to +1006
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think that we want this restriction to be more general and prevent using any of the RESERVED_ATTR_REL_HIERARCHICAL_NAMES names on any attribute or relationship that is defined by the user, not just on hierarchical nodes

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If we stick to restricting RESERVED_ATTR_REL_HIERARCHICAL_NAMES globally that would be the easiest solution as we can just iterate over the input from the user and restrict these attribute or relationship names from entering the system. My main concern would be if there are users out there that have already used these names anywhere.

I.e. today this is a perfectly valid schema:

# yaml-language-server: $schema=https://schema.infrahub.app/infrahub/schema/latest.json
---
version: '1.0'

nodes:
  - name: Person
    namespace: Test
    label: "Person"
    default_filter: name__value
    display_labels:
      - name__value
    attributes:
      - name: name
        kind: Text
      - name: children
        kind: Number
      - name: description
        kind: Text
        optional: true

It's fine to use "children" as the name of an attribute as long as the node itself isn't hierarchical. It might be perfectly fine but it could also be that some installations face problems due to this.


for attr in node.attributes:
if attr.name in RESERVED_ATTR_REL_HIERARCHICAL_NAMES:
raise ValueError(
f"{node.kind}: {attr.name} isn't allowed as an attribute name on hierarchical nodes."
)

for rel in node.relationships:
if rel.loaded_from_api and rel.name in RESERVED_ATTR_REL_HIERARCHICAL_NAMES:
raise ValueError(
f"{node.kind}: {rel.name} isn't allowed as a relationship name on hierarchical nodes."
)

def validate_python_keywords(self) -> None:
"""Validate that attribute and relationship names don't use Python keywords."""
for name in self.all_names:
Expand Down
64 changes: 64 additions & 0 deletions backend/tests/fixtures/schemas/restricted_names_01.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
{
"version": "1.0",
"generics": [
{
"name": "Location",
"namespace": "Generic",
"hierarchical": true,
"attributes": [
{
"name": "name_location",
"unique": true,
"optional": false,
"kind": "Text"
}
]
}
],
"nodes": [
{
"name": "Site",
"namespace": "Location",
"inherit_from": ["GenericLocation"],
"children": "TestingParent",
"parent": ""
},
{
"name": "Parent",
"namespace": "Testing",
"inherit_from": ["GenericLocation"],
"children": "",
"parent": "LocationSite",
"relationships": [
{
"name": "children",
"kind": "Generic",
"optional": true,
"peer": "TestingChild",
"cardinality": "many"
}
]
},
{
"name": "Child",
"namespace": "Testing",
"attributes": [
{
"name": "name_child",
"unique": true,
"optional": false,
"kind": "Text"
}
],
"relationships": [
{
"name": "relationship",
"kind": "Attribute",
"optional": false,
"peer": "TestingParent",
"cardinality": "one"
}
]
}
]
}
65 changes: 65 additions & 0 deletions backend/tests/fixtures/schemas/restricted_names_02.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
{
"version": "1.0",
"generics": [
{
"name": "Location",
"namespace": "Generic",
"hierarchical": true,
"attributes": [
{
"name": "name_location",
"unique": true,
"optional": false,
"kind": "Text"
}
]
}
],
"nodes": [
{
"name": "Site",
"namespace": "Location",
"inherit_from": ["GenericLocation"],
"children": "TestingParent",
"parent": ""
},
{
"name": "Parent",
"namespace": "Testing",
"inherit_from": ["GenericLocation"],
"children": "",
"parent": "LocationSite",
"attributes": [{"name": "children", "unique": true, "optional": false, "kind": "Text"}],
"relationships": [
{
"name": "parent_relationship",
"kind": "Generic",
"optional": true,
"peer": "TestingChild",
"cardinality": "many"
}
]
},
{
"name": "Child",
"namespace": "Testing",
"attributes": [
{
"name": "name_child",
"unique": true,
"optional": false,
"kind": "Text"
}
],
"relationships": [
{
"name": "child_relationship",
"kind": "Attribute",
"optional": false,
"peer": "TestingParent",
"cardinality": "one"
}
]
}
]
}
48 changes: 47 additions & 1 deletion backend/tests/unit/api/test_40_schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from infrahub import config
from infrahub.core import registry
from infrahub.core.branch import Branch
from infrahub.core.constants import InfrahubKind
from infrahub.core.constants import RESERVED_ATTR_REL_HIERARCHICAL_NAMES, InfrahubKind
from infrahub.core.initialization import create_branch
from infrahub.core.node import Node
from infrahub.core.schema import SchemaRoot, core_models
Expand Down Expand Up @@ -443,3 +443,49 @@ async def test_schema_read_endpoints_anonymous_account(
response = client.get("/api/schema/json_schema/TestCar")

assert response.status_code == 200 if allow_anonymous_access else 401


@pytest.mark.parametrize(
"reserved_name",
[pytest.param(reserved_name, id=reserved_name) for reserved_name in RESERVED_ATTR_REL_HIERARCHICAL_NAMES],
)
async def test_schema_load_restricted_names(
db: InfrahubDatabase,
client: TestClient,
admin_headers,
default_branch: Branch,
prefect_test_fixture,
workflow_local,
authentication_base,
helper,
reserved_name,
):
schema = helper.schema_file("restricted_names_01.json")
schema["nodes"][1]["relationships"][0]["name"] = reserved_name
with client:
response = client.post(
"/api/schema/load",
headers=admin_headers,
json={"schemas": [schema]},
)

assert response.status_code == 422
assert (
response.json()["errors"][0]["message"]
== f"TestingParent: {reserved_name} isn't allowed as a relationship name on hierarchical nodes."
)
Comment on lines +448 to +476
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🛠️ Refactor suggestion | 🟠 Major

Add a docstring to the test function.

The test logic and assertions are correct, but the function is missing a required docstring. All Python functions must include a docstring in Google-style format.

As per coding guidelines.

Apply this diff to add a docstring:

 @pytest.mark.parametrize(
     "reserved_name",
     [pytest.param(reserved_name, id=reserved_name) for reserved_name in RESERVED_ATTR_REL_HIERARCHICAL_NAMES],
 )
 async def test_schema_load_restricted_names(
     db: InfrahubDatabase,
     client: TestClient,
     admin_headers,
     default_branch: Branch,
     prefect_test_fixture,
     workflow_local,
     authentication_base,
     helper,
     reserved_name,
 ):
+    """Test that loading a schema with reserved hierarchical names fails with proper error.
+    
+    Verifies that relationship names matching RESERVED_ATTR_REL_HIERARCHICAL_NAMES
+    (parent, children, ancestors, descendants) are rejected for hierarchical nodes
+    when loading schemas via the API.
+    """
     schema = helper.schema_file("restricted_names_01.json")
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
@pytest.mark.parametrize(
"reserved_name",
[pytest.param(reserved_name, id=reserved_name) for reserved_name in RESERVED_ATTR_REL_HIERARCHICAL_NAMES],
)
async def test_schema_load_restricted_names(
db: InfrahubDatabase,
client: TestClient,
admin_headers,
default_branch: Branch,
prefect_test_fixture,
workflow_local,
authentication_base,
helper,
reserved_name,
):
schema = helper.schema_file("restricted_names_01.json")
schema["nodes"][1]["relationships"][0]["name"] = reserved_name
with client:
response = client.post(
"/api/schema/load",
headers=admin_headers,
json={"schemas": [schema]},
)
assert response.status_code == 422
assert (
response.json()["errors"][0]["message"]
== f"TestingParent: {reserved_name} isn't allowed as a relationship name on hierarchical nodes."
)
@pytest.mark.parametrize(
"reserved_name",
[pytest.param(reserved_name, id=reserved_name) for reserved_name in RESERVED_ATTR_REL_HIERARCHICAL_NAMES],
)
async def test_schema_load_restricted_names(
db: InfrahubDatabase,
client: TestClient,
admin_headers,
default_branch: Branch,
prefect_test_fixture,
workflow_local,
authentication_base,
helper,
reserved_name,
):
"""Test that loading a schema with reserved hierarchical names fails with proper error.
Verifies that relationship names matching RESERVED_ATTR_REL_HIERARCHICAL_NAMES
(parent, children, ancestors, descendants) are rejected for hierarchical nodes
when loading schemas via the API.
"""
schema = helper.schema_file("restricted_names_01.json")
schema["nodes"][1]["relationships"][0]["name"] = reserved_name
with client:
response = client.post(
"/api/schema/load",
headers=admin_headers,
json={"schemas": [schema]},
)
assert response.status_code == 422
assert (
response.json()["errors"][0]["message"]
== f"TestingParent: {reserved_name} isn't allowed as a relationship name on hierarchical nodes."
)
🤖 Prompt for AI Agents
In backend/tests/unit/api/test_40_schema.py around lines 448 to 476, the test
function test_schema_load_restricted_names is missing the required Google-style
docstring; add a docstring as the first statement in the function that
succinctly describes what the test verifies (that loading a schema with reserved
relationship names on hierarchical nodes returns a 422 with the expected error),
and include a brief Google-style Args section listing key fixtures/parameters
used by the test (e.g., db, client, admin_headers, default_branch, helper,
reserved_name).


schema = helper.schema_file("restricted_names_02.json")
schema["nodes"][1]["attributes"][0]["name"] = reserved_name
with client:
response = client.post(
"/api/schema/load",
headers=admin_headers,
json={"schemas": [schema]},
)

assert response.status_code == 422
assert (
response.json()["errors"][0]["message"]
== f"TestingParent: {reserved_name} isn't allowed as an attribute name on hierarchical nodes."
)
Loading