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
6 changes: 5 additions & 1 deletion infrahub_sdk/template/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@
JinjaTemplateSyntaxError,
JinjaTemplateUndefinedError,
)
from .filters import AVAILABLE_FILTERS
from .filters import AVAILABLE_FILTERS, INFRAHUB_FILTERS
from .models import UndefinedJinja2Error

netutils_filters = jinja2_convenience_function()
Expand Down Expand Up @@ -155,6 +155,10 @@ def _set_filters(self, env: jinja2.Environment) -> None:
env.filters.update(
{name: jinja_filter for name, jinja_filter in netutils_filters.items() if name in self._available_filters}
)
# Add filters from our own SDK
env.filters.update(
{name: jinja_filter for name, jinja_filter in INFRAHUB_FILTERS.items() if name in self._available_filters}
)
# Add user supplied filters
env.filters.update(self._filters)

Expand Down
51 changes: 50 additions & 1 deletion infrahub_sdk/template/filters.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,7 @@
from dataclasses import dataclass
from enum import Enum
from importlib import import_module
from typing import Any


@dataclass
Expand All @@ -8,6 +11,52 @@ class FilterDefinition:
source: str


def value_to_enum_name(value: Any, enum_path: str | None = None) -> str:
"""Convert a value to its enum member name using the specified enum class.

This filter takes a raw value and converts it to the corresponding enum member name by dynamically importing the
enum class.

For example, `{{ decision__value | value_to_enum_name("infrahub.core.constants.PermissionDecision") }}`
Copy link
Contributor

Choose a reason for hiding this comment

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

Not sure out the format of this filter, it feels a bit complicated to use for the end user

Copy link
Contributor

Choose a reason for hiding this comment

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

I don't know if it will be useful at all to end users. I think this is pretty much a workaround for https://github.com/opsmill/infrahub/blob/infrahub-v1.6.2/backend/infrahub/core/node/permissions.py#L34 so that we can clean up the "identifier" fields for computed attributes and make them into a computed attribute instead. As such I'm not sure if we should have this in the SDK or if it will just confuse users. Of course after saying that there's little value for end users to have this we could get a feature request for it tomorrow, but given the problem it sets out to solve I think it's fine that it might be a bit complicated to use.

will return: `"ALLOW_ALL"` for value `6`.
"""
if isinstance(value, Enum) and not enum_path:
return value.name

raw_value = value.value if isinstance(value, Enum) else value

if not enum_path:
return str(raw_value)

try:
module_path, class_name = enum_path.rsplit(".", 1)
module = import_module(module_path)
enum_type = getattr(module, class_name)
Comment on lines +31 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.

is allowing the user to import any Python module inside of the SDK a security concern?
I guess the user can already do this b/c they are using the SDK in their own code, but we also use the SDK within Infrahub and I wonder if there is a way for the user to force the Infrahub server to import something malicious

Copy link
Contributor Author

Choose a reason for hiding this comment

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

We use Jinja2's SandboxedEnvironment to render template, so I wonder if this could help prevent having malicious code messing around (I would not expect it to be 100% secure though). That said we also allow users to import their own Jinja2 filters so I guess these could also be entry points to run any kind of code.

Copy link
Contributor

Choose a reason for hiding this comment

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

This shouldn't be executed in any environment where the user can choose their own code to import and not already have that ability. Based on that I think it should be fine to do it this way.

except (ValueError, ImportError, AttributeError) as exc:
msg = f"Failed to resolve enum '{enum_path}': {exc}"
raise ValueError(msg) from exc

# Verify that we have a class and that this class is a valid Enum
if not (isinstance(enum_type, type) and issubclass(enum_type, Enum)):
raise ValueError(f"Resolved type '{enum_path}' is not a valid Enum")

try:
enum_member = enum_type(raw_value)
if enum_member.name is not None:
return enum_member.name
except (ValueError, TypeError) as exc:
msg = f"Value '{raw_value}' not found in enum '{enum_path}': {exc}"
raise ValueError(msg) from exc

return str(raw_value)
Comment on lines +43 to +51
Copy link

Choose a reason for hiding this comment

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

⚠️ Potential issue | 🟡 Minor

Remove unreachable code.

Line 51 is unreachable because the try block (lines 43-49) will always either return on line 46 or raise on line 49. Additionally, the None check on line 45 is unnecessary since enum member names are never None.

🔎 Proposed fix
     try:
         enum_member = enum_type(raw_value)
-        if enum_member.name is not None:
-            return enum_member.name
+        return enum_member.name
     except (ValueError, TypeError) as exc:
         msg = f"Value '{raw_value}' not found in enum '{enum_path}': {exc}"
         raise ValueError(msg) from exc
-
-    return str(raw_value)
🤖 Prompt for AI Agents
In infrahub_sdk/template/filters.py around lines 43-51, remove the unreachable
None-check and trailing return: instantiate enum_member = enum_type(raw_value)
inside the try and immediately return enum_member.name (no need to check for
None), keep the except block as-is to raise a ValueError with details, and
delete the final return str(raw_value) since it is unreachable.



INFRAHUB_FILTERS = {"value_to_enum_name": value_to_enum_name}
INFRAHUB_FILTER_DEFINITIONS = [
FilterDefinition(name=name, trusted=True, source="infrahub-sdk-python") for name in sorted(INFRAHUB_FILTERS.keys())
]
Comment on lines +55 to +57
Copy link
Contributor Author

Choose a reason for hiding this comment

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

I wrote it like that so that we can easily add filters in the future.



BUILTIN_FILTERS = [
FilterDefinition(name="abs", trusted=True, source="jinja2"),
FilterDefinition(name="attr", trusted=False, source="jinja2"),
Expand Down Expand Up @@ -148,4 +197,4 @@ class FilterDefinition:
]


AVAILABLE_FILTERS = BUILTIN_FILTERS + NETUTILS_FILTERS
AVAILABLE_FILTERS = BUILTIN_FILTERS + NETUTILS_FILTERS + INFRAHUB_FILTER_DEFINITIONS
112 changes: 112 additions & 0 deletions tests/unit/sdk/test_template.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from dataclasses import dataclass, field
from enum import IntEnum
from pathlib import Path
from typing import Any

Expand All @@ -16,8 +17,10 @@
)
from infrahub_sdk.template.filters import (
BUILTIN_FILTERS,
INFRAHUB_FILTER_DEFINITIONS,
NETUTILS_FILTERS,
FilterDefinition,
value_to_enum_name,
)
from infrahub_sdk.template.models import UndefinedJinja2Error

Expand Down Expand Up @@ -310,3 +313,112 @@ def _compare_errors(expected: JinjaTemplateError, received: JinjaTemplateError)

else:
raise Exception("This should never happen")


class SampleIntEnum(IntEnum):
DENY = 1
ALLOW_DEFAULT = 2
ALLOW_OTHER = 4
ALLOW_ALL = 6


TEST_ENUM_PATH = "tests.unit.sdk.test_template.SampleIntEnum"


def test_validate_infrahub_filter_sorting() -> None:
"""Test to validate that infrahub-sdk-python filter names are in alphabetical order."""
names = [filter_definition.name for filter_definition in INFRAHUB_FILTER_DEFINITIONS]
assert names == sorted(names)


@pytest.mark.parametrize(
("value", "enum_path", "expected"),
[
pytest.param(1, TEST_ENUM_PATH, "DENY", id="int-value-deny"),
pytest.param(6, TEST_ENUM_PATH, "ALLOW_ALL", id="int-value-allow-all"),
pytest.param(SampleIntEnum.DENY, TEST_ENUM_PATH, "DENY", id="enum-input-deny"),
pytest.param(SampleIntEnum.ALLOW_ALL, TEST_ENUM_PATH, "ALLOW_ALL", id="enum-input-allow-all"),
pytest.param(6, None, "6", id="no-enum-path-int"),
pytest.param("test", None, "test", id="no-enum-path-str"),
],
)
def test_value_to_enum_name_success(value: int | str | SampleIntEnum, enum_path: str | None, expected: str) -> None:
assert value_to_enum_name(value, enum_path) == expected


@pytest.mark.parametrize(
("value", "enum_path", "error_pattern"),
[
pytest.param(
6,
"nonexistent.module.EnumClass",
r"Failed to resolve enum 'nonexistent\.module\.EnumClass'",
id="invalid-module",
),
pytest.param(6, "enum.NonExistentEnum", r"Failed to resolve enum 'enum\.NonExistentEnum'", id="invalid-class"),
pytest.param(
"invalid", TEST_ENUM_PATH, f"Value 'invalid' not found in enum '{TEST_ENUM_PATH}'", id="invalid-value"
),
pytest.param(0, TEST_ENUM_PATH, f"Value '0' not found in enum '{TEST_ENUM_PATH}'", id="zero-not-in-enum"),
pytest.param(6, "NoDotInPath", "Failed to resolve enum 'NoDotInPath'", id="invalid-path-format"),
pytest.param(
6,
"dataclasses.dataclass",
r"Resolved type 'dataclasses\.dataclass' is not a valid Enum",
id="non-enum-class",
),
],
)
def test_value_to_enum_name_errors(value: int | str, enum_path: str, error_pattern: str) -> None:
with pytest.raises(ValueError, match=error_pattern):
value_to_enum_name(value, enum_path)


VALUE_TO_ENUM_NAME_FILTER_TEST_CASES = [
JinjaTestCase(
name="value-to-enum-name-with-full-path-deny",
template="{{ decision | value_to_enum_name('" + TEST_ENUM_PATH + "') }}",
variables={"decision": 1},
expected="DENY",
expected_variables=["decision"],
),
JinjaTestCase(
name="value-to-enum-name-with-full-path-allow-all",
template="{{ decision | value_to_enum_name('" + TEST_ENUM_PATH + "') }}",
variables={"decision": 6},
expected="ALLOW_ALL",
expected_variables=["decision"],
),
JinjaTestCase(
name="value-to-enum-name-global-permission-format",
template="global:{{ action }}:{{ decision | value_to_enum_name('" + TEST_ENUM_PATH + "') | lower }}",
variables={"action": "manage_accounts", "decision": 6},
expected="global:manage_accounts:allow_all",
expected_variables=["action", "decision"],
),
JinjaTestCase(
name="value-to-enum-name-object-permission-format",
template="object:{{ ns }}:{{ nm }}:{{ action }}:{{ decision | value_to_enum_name('"
+ TEST_ENUM_PATH
+ "') | lower }}",
variables={"ns": "Infra", "nm": "Device", "action": "view", "decision": 2},
expected="object:Infra:Device:view:allow_default",
expected_variables=["action", "decision", "nm", "ns"],
),
]


@pytest.mark.parametrize(
"test_case",
[pytest.param(tc, id=tc.name) for tc in VALUE_TO_ENUM_NAME_FILTER_TEST_CASES],
)
async def test_value_to_enum_name_filter_in_templates(test_case: JinjaTestCase) -> None:
jinja = Jinja2Template(template=test_case.template)
assert test_case.expected == await jinja.render(variables=test_case.variables)
assert test_case.expected_variables == jinja.get_variables()


async def test_value_to_enum_name_filter_in_templates_with_invalid_path() -> None:
jinja = Jinja2Template(template="{{ decision | value_to_enum_name('invalid.path.Enum') }}")
with pytest.raises(JinjaTemplateError):
await jinja.render(variables={"decision": 6})