Skip to content
Draft
Show file tree
Hide file tree
Changes from 4 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") }}`
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.

Copy link
Contributor

Choose a reason for hiding this comment

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

Let's not include that in the SDK then, let's create a dedicated filter for the permission enum then.

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 +33 to +36
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):
return str(raw_value)
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 get an import error I think it would be better to have a proper failure. Potentially there would be some future migration issues if we remove a decision from the permissions and the old value no longer exists in the enum, but that should preferably be handled by a migration. So it could be that by catching all of these errors and returning some value we mask a user error if they have made one. Where if we fail early the error should be logged somewhere.


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