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
49 changes: 48 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,50 @@ 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") }}`
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

except (ValueError, ImportError, AttributeError):
return str(raw_value)

# 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)):
return str(raw_value)

try:
enum_member = enum_type(raw_value)
if enum_member.name is not None:
return enum_member.name
except (ValueError, TypeError):
pass

return str(raw_value)


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 +53 to +55
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 +195,4 @@ class FilterDefinition:
]


AVAILABLE_FILTERS = BUILTIN_FILTERS + NETUTILS_FILTERS
AVAILABLE_FILTERS = BUILTIN_FILTERS + NETUTILS_FILTERS + INFRAHUB_FILTER_DEFINITIONS
109 changes: 109 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,109 @@ 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)


def test_value_to_enum_name_with_full_import_path() -> None:
for enum_entry in SampleIntEnum:
assert value_to_enum_name(enum_entry.value, TEST_ENUM_PATH) == enum_entry.name


def test_value_to_enum_name_with_enum_input() -> None:
assert value_to_enum_name(SampleIntEnum.DENY, TEST_ENUM_PATH) == "DENY"
assert value_to_enum_name(SampleIntEnum.ALLOW_ALL, TEST_ENUM_PATH) == "ALLOW_ALL"


def test_value_to_enum_name_without_enum_path() -> None:
assert value_to_enum_name(6) == "6"
assert value_to_enum_name("test") == "test"


def test_value_to_enum_name_with_invalid_module() -> None:
assert value_to_enum_name(6, "nonexistent.module.EnumClass") == "6"


def test_value_to_enum_name_with_invalid_class() -> None:
assert value_to_enum_name(6, "enum.NonExistentEnum") == "6"


def test_value_to_enum_name_with_non_enum_class() -> None:
assert value_to_enum_name(6, "dataclasses.dataclass") == "6"


def test_value_to_enum_name_with_invalid_value() -> None:
assert value_to_enum_name("invalid", TEST_ENUM_PATH) == "invalid"


def test_value_to_enum_name_with_zero() -> None:
assert value_to_enum_name(0, TEST_ENUM_PATH) == "0"


def test_value_to_enum_name_with_invalid_path_format() -> None:
assert value_to_enum_name(6, "NoDotInPath") == "6"


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"],
),
JinjaTestCase(
name="value-to-enum-name-with-invalid-path",
template="{{ decision | value_to_enum_name('invalid.path.Enum') }}",
variables={"decision": 6},
expected="6",
expected_variables=["decision"],
),
]


@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()