{% include "network.lua.j2" %}
-- VIEWS section ------------------------------------
-{% include "views.lua.j2" %}
+{# {% include "views.lua.j2" %} #}
-- LOCAL-DATA section -----------------------------
{% include "local_data.lua.j2" %}
-- SLICES section -----------------------------------
--- {% include "slices.lua.j2" %}
+{# {% include "slices.lua.j2" %} #}
-- POLICY section -----------------------------------
--- {% include "policy.lua.j2" %}
+{# {% include "policy.lua.j2" %} #}
-- RPZ section --------------------------------------
--- {% include "rpz.lua.j2" %}
+{# {% include "rpz.lua.j2" %} #}
-- FORWARD section ----------------------------
{% include "forward.lua.j2" %}
-{% macro view_tsig(tsig, rule) -%}
-view:tsig('{{ tsig }}',{{ rule }})
+{% macro view_insert_action(subnet, action) -%}
+assert(C.kr_view_insert_action('{{ subnet }}',{{ action }})==0)
{%- endmacro %}
-{% macro view_addr(addr, rule) -%}
-view:addr('{{ addr }}',{{ rule }})
-{%- endmacro %}
\ No newline at end of file
+
+{% macro view_options_flags(options) -%}
+policy.FLAGS({
+{%- if not options.minimize -%}
+'NO_MINIMIZE',
+{%- endif -%}
+{%- if not options.dns64 -%}
+'DNS64_DISABLE',
+{%- endif -%}
+})
+{%- endmacro %}
+
+
+{% macro view_answer(answer) -%}
+{%- if answer == 'allow' -%}
+policy.TAGS_ASSIGN({})
+{%- elif answer == 'refused' -%}
+policy.REFUSE
+{%- elif answer == 'noanswer' -%}
+policy.NO_ANSWER
+{%- endif -%}
+{%- endmacro %}
-{% if cfg.views %}
-modules.load('view')
-{% endif %}
\ No newline at end of file
+{% from 'macros/view_macros.lua.j2' import view_insert_action, view_options_flags, view_answer %}
+{% from 'macros/policy_macros.lua.j2' import policy_tags_assign %}
+
+local C = require('ffi').C
+
+{% for view in cfg.views %}
+{% for subnet in view.subnets %}
+
+{% if view.tags -%}
+{{ view_insert_action(subnet, policy_tags_assign(view.tags)) }}
+{% elif view.answer -%}
+{{ view_insert_action(subnet, view_answer(view.answer)) }}
+{%- endif %}
+
+{{ view_insert_action(subnet, view_options_flags(view.options)) }}
+
+{% endfor %}
+{% endfor %}
---
subnets: Identifies the client based on his subnet.
tags: Tags to link with other policy rules.
- options: Configuration options for clients identified by the view.
answer: Direct approach how to handle request from clients identified by the view.
+ options: Configuration options for clients identified by the view.
"""
- subnets: Optional[Union[List[IPNetwork], IPNetwork]] = None
+ subnets: Optional[Union[List[IPNetwork], IPNetwork]]
tags: Optional[List[IDPattern]] = None
+ answer: Optional[Literal["allow", "refused", "noanswer"]] = None
options: ViewOptionsSchema = ViewOptionsSchema()
- answer: Optional[Literal["allow", "refused"]] = None
+
+ def _validate(self) -> None:
+ if bool(self.tags) == bool(self.answer):
+ raise ValueError("only one of 'tags' and 'answer' options must be configured")
+from typing import Any
+
+import pytest
+
from knot_resolver_manager.datamodel.config_schema import template_from_str
-from knot_resolver_manager.datamodel.types import IPAddressOptionalPort
+from knot_resolver_manager.datamodel.view_schema import ViewOptionsSchema, ViewSchema
+
+
+def test_view_insert_action():
+ subnet = "10.0.0.0/8"
+ action = "policy.DENY"
+ tmpl_str = """{% from 'macros/view_macros.lua.j2' import view_insert_action %}
+{{ view_insert_action(subnet, action) }}"""
+
+ tmpl = template_from_str(tmpl_str)
+ assert tmpl.render(subnet=subnet, action=action) == f"assert(C.kr_view_insert_action('{ subnet }',{ action })==0)"
+
+
+def test_view_options_flags():
+ tmpl_str = """{% from 'macros/view_macros.lua.j2' import view_options_flags %}
+{{ view_options_flags(options) }}"""
+
+ tmpl = template_from_str(tmpl_str)
+ options = ViewOptionsSchema({"dns64": False, "minimize": False})
+ assert tmpl.render(options=options) == "policy.FLAGS({'NO_MINIMIZE','DNS64_DISABLE',})"
+ assert tmpl.render(options=ViewOptionsSchema()) == "policy.FLAGS({})"
-def test_view_tsig():
- tsig: str = r"\5mykey"
- rule = "policy.all(policy.DENY)"
- tmpl_str = """{% from 'macros/view_macros.lua.j2' import view_tsig %}
-{{ view_tsig(tsig, rule) }}"""
+def test_view_answer():
+ tmpl_str = """{% from 'macros/view_macros.lua.j2' import view_options_flags %}
+{{ view_options_flags(options) }}"""
tmpl = template_from_str(tmpl_str)
- assert tmpl.render(tsig=tsig, rule=rule) == f"view:tsig('{tsig}',{rule})"
+ options = ViewOptionsSchema({"dns64": False, "minimize": False})
+ assert tmpl.render(options=options) == "policy.FLAGS({'NO_MINIMIZE','DNS64_DISABLE',})"
+ assert tmpl.render(options=ViewOptionsSchema()) == "policy.FLAGS({})"
-def test_view_addr():
- addr: IPAddressOptionalPort = IPAddressOptionalPort("10.0.0.1")
- rule = "policy.all(policy.DENY)"
- tmpl_str = """{% from 'macros/view_macros.lua.j2' import view_addr %}
-{{ view_addr(addr, rule) }}"""
+@pytest.mark.parametrize(
+ "val,res",
+ [
+ ("allow", "policy.TAGS_ASSIGN({})"),
+ ("refused", "policy.REFUSE"),
+ ("noanswer", "policy.NO_ANSWER"),
+ ],
+)
+def test_view_answer(val: Any, res: Any):
+ tmpl_str = """{% from 'macros/view_macros.lua.j2' import view_answer %}
+{{ view_answer(view.answer) }}"""
tmpl = template_from_str(tmpl_str)
- assert tmpl.render(addr=addr, rule=rule) == f"view:addr('{addr}',{rule})"
+ view = ViewSchema({"subnets": "10.0.0.0/8", "answer": val})
+ assert tmpl.render(view=view) == res
projects / thirdparty / knot-resolver.git / commitdiff
