nethsec.dpi

package documentation
(source)

Library that handles the DPI rules.

From __init__.py:

Function add_appgroup Store a new application group.
Function add_rule Store a new rule.
Function build_rule_criteria Build the criteria of a managed rule: the source, when set, ANDed with the union of its groups.
Function delete_appgroup Delete an application group.
Function delete_rule Delete a rule and close the gap its priority leaves behind.
Function disable_rule Disable a rule, managed or not.
Function edit_appgroup Edit an application group.
Function edit_rule Edit a rule. Only rules created through the API can be edited: a rule carrying a hand-written criteria has no source and no group to fill the form with, and rewriting it would change what it matches.
Function enable_rule Enable a rule, managed or not.
Function expand_appgroups Expand application groups into the parenthesised OR block of an action criteria.
Function expand_source Expand the source of a rule into values a criteria can match: ranges become CIDR blocks, addresses and networks are left as they are.
Function freeze_legacy_criteria Build the criteria of a rule written before application groups existed, the way dpi-config has always generated it: source, application, protocol and category turned into an expression narrowed to the rule's device, with the VLAN rewrite applied when the device names a VLAN...
Function is_used_appgroup Check if an application group is referenced by a rule.
Function list_appgroup_catalog List every application and protocol a group can be built with, grouped by category.
Function list_appgroups List the application groups, ordered by name.
Function list_rules List the rules in priority order, the order they are evaluated in.
Function load_applications Reads the applications loaded in memory by the engine, using the netifyd --dump-apps command.
Function load_categories Reads the categories loaded in memory by the engine, using the netifyd --dump-categories command.
Function load_protocols Reads the protocols from the netifyd --dump-protos command.
Function migrate_schema Migrate every rule and exemption from the schema used before application groups existed.
Function order_rules Reorder every rule, renumbering the priorities densely from the given order.
Function rename_rule Rename a rule, managed or not.
Constant APPGROUP_CATALOG_FILES Undocumented
Constant APPGROUP_MEMBERS Undocumented
Constant APPGROUP_NAME_MAX_LENGTH Undocumented
Constant DPI_MATCH_ALL_CRITERIA Undocumented
Constant DPI_RULE_ACTIONS Undocumented
Constant DPI_RULE_NAME_MAX_LENGTH Undocumented
Constant DPI_RULE_POSITIONS Undocumented
Constant NETIFYD_DATA_DIR Undocumented
Constant UNCATEGORIZED_TAG Undocumented
Function __appgroup_vocabulary Values a member of the given kind can take, lowercased.
Function __exemption_source Turn a global exemption's criteria into a rule source. A firewall object is expanded into its member addresses; a plain address, network or range is kept as is.
Function __is_rule Undocumented
Function __load_catalog Reads a downloaded catalog, one of the /etc/netifyd/netify-*.json files.
Function __renumber_rules Give every rule a dense priority, starting at 1. The generator keeps 0 for its own action, so no user rule can be evaluated before it.
Function __save_appgroup_data Undocumented
Function __save_rule_data Undocumented
Function __sorted_rules Rule sections in priority order. Rules with no priority, i.e. written by hand, go last keeping their config order.
Function __validate_appgroup Validate a group before it is stored.
Function __validate_appgroup_members Validate the members of one kind, stripping and deduplicating them.
Function __validate_appgroups Validate the groups a rule matches. At least one is required: a rule with a source and no group would be an IP-level rule, which the firewall does better, and a rule with neither would expand to an empty criteria.
Function __validate_rule_name Undocumented
Function __validate_source Validate the source of a rule: addresses, networks and ranges, IPv4 and IPv6 alike. An empty source means the rule matches every host.
Constant __APPGROUP_FORBIDDEN_CHARS Undocumented
Constant __APPGROUP_MEMBER_ERRORS Undocumented
def add_appgroup(e_uci: EUci, name: str, applications: list[str] = None, application_categories: list[str] = None, protocols: list[str] = None, protocol_categories: list[str] = None) -> str: (source)

Store a new application group.

Args:
  • e_uci: euci instance
  • name: name of the group
  • applications: list of application names, as the engine reports them
  • application_categories: list of application category tags
  • protocols: list of protocol names, as the engine reports them
  • protocol_categories: list of protocol category tags
Returns:
config name of the group created
Raises:
  • ValidationError: if the name or any member is invalid, or if the group has no members
def add_rule(e_uci: EUci, name: str, enabled: bool, action: str, source: list[str], appgroups: list[str], position: str = 'bottom', match_all: bool = False) -> str: (source)

Store a new rule.

Args:
  • e_uci: euci instance
  • name: name of the rule
  • enabled: enable the rule
  • action: 'block' or 'allow'
  • source: list of addresses, networks or ranges, empty to match every host
  • appgroups: config names of the application groups the rule matches, at least one unless match_all is set
  • position: 'top' to evaluate the rule before every other one, 'bottom' after them
  • match_all: the rule matches every flow, narrowed to source when one is given. Mutually exclusive with appgroups
Returns:
config name of the rule created
Raises:
  • ValidationError: if any argument is invalid
def build_rule_criteria(e_uci: EUci, rule: dict) -> str: (source)

Build the criteria of a managed rule: the source, when set, ANDed with the union of its groups.

Args:
  • e_uci: euci instance
  • rule: the rule section, as returned by uci.get_all
Returns:
the expression, or an empty string when the rule matches nothing and must not be emitted
def delete_appgroup(e_uci: EUci, config_name: str) -> str: (source)

Delete an application group.

Args:
  • e_uci: euci instance
  • config_name: config name of the group to delete
Returns:
config name of the group deleted
Raises:
  • ValidationError: if the group does not exist or is referenced by a rule
def delete_rule(e_uci: EUci, config_name: str) -> str: (source)

Delete a rule and close the gap its priority leaves behind.

Args:
  • e_uci: euci instance
  • config_name: config name of the rule to delete
Returns:
config name of the rule deleted
Raises:
  • ValidationError: if the rule does not exist
def disable_rule(e_uci: EUci, config_name: str) -> str: (source)

Disable a rule, managed or not.

Args:
  • e_uci: euci instance
  • config_name: config name of the rule to disable
Returns:
config name of the rule disabled
Raises:
  • ValidationError: if the rule does not exist
def edit_appgroup(e_uci: EUci, config_name: str, name: str, applications: list[str] = None, application_categories: list[str] = None, protocols: list[str] = None, protocol_categories: list[str] = None) -> str: (source)

Edit an application group.

Args:
  • e_uci: euci instance
  • config_name: config name of the group to edit
  • name: name of the group
  • applications: list of application names, as the engine reports them
  • application_categories: list of application category tags
  • protocols: list of protocol names, as the engine reports them
  • protocol_categories: list of protocol category tags
Returns:
config name of the group edited
Raises:
  • ValidationError: if the group does not exist, or if the name or any member is invalid
def edit_rule(e_uci: EUci, config_name: str, name: str, enabled: bool, action: str, source: list[str], appgroups: list[str], match_all: bool = False) -> str: (source)

Edit a rule. Only rules created through the API can be edited: a rule carrying a hand-written criteria has no source and no group to fill the form with, and rewriting it would change what it matches.

Args:
  • e_uci: euci instance
  • config_name: config name of the rule to edit
  • name: name of the rule
  • enabled: enable the rule
  • action: 'block' or 'allow'
  • source: list of addresses, networks or ranges, empty to match every host
  • appgroups: config names of the application groups the rule matches, at least one unless match_all is set
  • match_all: the rule matches every flow, narrowed to source when one is given. Mutually exclusive with appgroups
Returns:
config name of the rule edited
Raises:
  • ValidationError: if the rule does not exist, is not managed, or any argument is invalid
def enable_rule(e_uci: EUci, config_name: str) -> str: (source)

Enable a rule, managed or not.

Args:
  • e_uci: euci instance
  • config_name: config name of the rule to enable
Returns:
config name of the rule enabled
Raises:
  • ValidationError: if the rule does not exist
def expand_appgroups(e_uci: EUci, config_names: list[str]) -> str: (source)

Expand application groups into the parenthesised OR block of an action criteria.

Members of several groups are merged and deduplicated; categories are never expanded into their applications, so a category keeps following the signature updates. Groups that do not exist are skipped.

Args:
  • e_uci: euci instance
  • config_names: config names of the groups to expand
Returns:
the expression, e.g. (app == 'netify.amazon' || proto == 'http/connect'), or an empty string if the groups hold no member
def expand_source(values: list[str]) -> list[str]: (source)

Expand the source of a rule into values a criteria can match: ranges become CIDR blocks, addresses and networks are left as they are.

Args:
  • values: source entries of the rule
Returns:
list of addresses and CIDR blocks
def freeze_legacy_criteria(e_uci: EUci, rule: dict) -> str: (source)

Build the criteria of a rule written before application groups existed, the way dpi-config has always generated it: source, application, protocol and category turned into an expression narrowed to the rule's device, with the VLAN rewrite applied when the device names a VLAN. Used by the migration to freeze a rule's behaviour verbatim before clearing those fields.

Args:
  • e_uci: euci instance
  • rule: the rule section, as returned by uci.get_all
Returns:
the semicolon-terminated expression
def is_used_appgroup(e_uci: EUci, config_name: str) -> tuple[bool, list[str]]: (source)

Check if an application group is referenced by a rule.

Args:
  • e_uci: euci instance
  • config_name: config name of the group
Returns:
A tuple with: - True if the group is referenced by at least one rule, False otherwise - a list of the rule sections referencing it
def list_appgroup_catalog() -> dict[str, list[dict]]: (source)

List every application and protocol a group can be built with, grouped by category.

Every source has to answer: the engine says what is matchable, the catalogs say how it reads, and a partial answer would silently offer a smaller product than the machine has. So this raises instead of returning what it managed to collect.

The id of an item is what a group stores, and it is the engine name whenever there is one: the catalog tag is only a fallback for an entry that cannot be matched anyway, so it is never a value a client submits. A group is selectable when the category itself can be a group member, which every real category can be and the uncategorised bucket never can. Groups and items are sorted by label, with the uncategorised bucket last; a client translating the category labels has to sort them again.

Returns:
dict with the keys "applications" and "protocols", each a list of category groups
Raises:
  • Exception: if the engine cannot be queried or a catalog is missing
def list_appgroups(e_uci: EUci, search: str = None, limit: int = None, page: int = 1, used_info: bool = True) -> dict: (source)

List the application groups, ordered by name.

Args:
  • e_uci: euci instance
  • search: search string, matched against the group name
  • limit: limit the number of results, all of them if not given
  • page: page number
  • used_info: include the used and matches info
Returns:
dict with the "data" list and the "meta" pagination info
def list_rules(e_uci: EUci) -> list[dict]: (source)

List the rules in priority order, the order they are evaluated in.

Rules hidden with ns_visible '0' are left out: they are system entries the user must not touch.

Args:
  • e_uci: euci instance
Returns:
list of dicts, each dict contains the property "id", "name", "enabled", "action", "source", "appgroups", "match_all", "managed" and "index", plus "criteria" for the rules the API did not create
def load_applications() -> dict[int, str]: (source)

Reads the applications loaded in memory by the engine, using the netifyd --dump-apps command.

Returns:
dict of applications, the key is the application id and the value is the application name
def load_categories() -> dict[str, dict[int, str]]: (source)

Reads the categories loaded in memory by the engine, using the netifyd --dump-categories command.

The dump reports one category per line, as "<id>: <type>: <tag>".

Returns:
dict of categories indexed by type, e.g. {'application': {1: 'adult'}, 'protocol': {2: 'database'}}. Ids are the ones the agent uses internally: they do not match the ids of the downloaded catalogs, so categories must always be crossed by tag.
def load_protocols() -> dict[int, str]: (source)

Reads the protocols from the netifyd --dump-protos command.

Returns:
dict of protocols, each dict contains the property "id" and "name"
def migrate_schema(e_uci: EUci): (source)

Migrate every rule and exemption from the schema used before application groups existed.

Each legacy rule (device, application, protocol or category set) is turned into an unmanaged rule with its behaviour frozen verbatim into criteria — see freeze_legacy_criteria. A rule whose action is a retired QoS value, or that ends up matching nothing, is dropped instead, with a log line naming it. A per-rule exemption or log (fields no recent API ever wrote) is removed from the rule, with a log line as well.

Each global exemption becomes an Allow rule at the top of the list: a managed match-all rule narrowed to a source when its criteria is a plain address, CIDR or firewall object; an unmanaged one carrying the criteria verbatim otherwise. Disabled exemptions become disabled rules. The rule is named after the exemption description, cut to the rule name limit, or Migrated exception N when it has none. The exemption section type and popular_filters are then removed.

The changes are committed and the dpi service reloaded right away.

Safe to call unconditionally on every boot: a box with nothing left in the old schema returns without touching UCI.

Args:
  • e_uci: euci instance
def order_rules(e_uci: EUci, order: list[str]) -> list[str]: (source)

Reorder every rule, renumbering the priorities densely from the given order.

The whole set must be listed, hidden rules included: a partial order would silently move the rules left out, which with a paginated list is a rule the caller never saw.

Args:
  • e_uci: euci instance
  • order: config names of every rule, in the order they must be evaluated
Returns:
the list of the ordered rules
Raises:
  • ValidationError: if the order does not name every rule exactly once
def rename_rule(e_uci: EUci, config_name: str, name: str) -> str: (source)

Rename a rule, managed or not.

Args:
  • e_uci: euci instance
  • config_name: config name of the rule to rename
  • name: new name of the rule
Returns:
config name of the rule renamed
Raises:
  • ValidationError: if the rule does not exist or the name is invalid
APPGROUP_CATALOG_FILES: dict[str, str] = (source)

Undocumented

Value
{'applications': 'netify-application-catalog.json',
 'protocols': 'netify-protocol-catalog.json'}
APPGROUP_MEMBERS: dict[str, str] = (source)

Undocumented

Value
{'app': 'applications',
 'app_category': 'application_categories',
 'proto': 'protocols',
 'proto_category': 'protocol_categories'}
APPGROUP_NAME_MAX_LENGTH: int = (source)

Undocumented

Value
64
DPI_MATCH_ALL_CRITERIA: str = (source)

Undocumented

Value
'*'
DPI_RULE_ACTIONS: tuple[str, ...] = (source)

Undocumented

Value
('block', 'allow')
DPI_RULE_NAME_MAX_LENGTH: int = (source)

Undocumented

Value
64
DPI_RULE_POSITIONS: tuple[str, ...] = (source)

Undocumented

Value
('top', 'bottom')
NETIFYD_DATA_DIR: str = (source)

Undocumented

Value
'/etc/netifyd'
UNCATEGORIZED_TAG: str = (source)

Undocumented

Value
''
def __appgroup_vocabulary(kind: str) -> set[str] | None: (source)

Values a member of the given kind can take, lowercased.

The engine is the authority: only what it has loaded can be matched. The downloaded catalog backs applications up, so that a group stays editable on a machine that lost its premium signatures.

Args:
Returns:
set of accepted values, or None when no source is available and the values cannot be checked
def __exemption_source(e_uci: EUci, criteria: str) -> list[str] | None: (source)

Turn a global exemption's criteria into a rule source. A firewall object is expanded into its member addresses; a plain address, network or range is kept as is.

Returns:
the source list, or None when the criteria is empty or does not resolve to at least one address, telling the caller to fall back to an unmanaged rule carrying it verbatim
def __is_rule(e_uci: EUci, config_name: str) -> bool: (source)

Undocumented

def __load_catalog(filename: str) -> list[dict]: (source)

Reads a downloaded catalog, one of the /etc/netifyd/netify-*.json files.

Args:
  • filename: name of the catalog file inside the netifyd data directory
Returns:
list of catalog entries
def __renumber_rules(e_uci: EUci, order: list[str] = None): (source)

Give every rule a dense priority, starting at 1. The generator keeps 0 for its own action, so no user rule can be evaluated before it.

def __save_appgroup_data(e_uci: EUci, config_name: str, name: str, members: dict[str, list[str]]): (source)

Undocumented

def __save_rule_data(e_uci: EUci, config_name: str, name: str, enabled: bool, action: str, source: list[str], appgroups: list[str], match_all: bool = False): (source)

Undocumented

def __sorted_rules(e_uci: EUci) -> list[tuple[str, dict]]: (source)

Rule sections in priority order. Rules with no priority, i.e. written by hand, go last keeping their config order.

def __validate_appgroup(e_uci: EUci, name: str, applications: list[str], application_categories: list[str], protocols: list[str], protocol_categories: list[str], config_name: str = None) -> tuple[str, dict[str, list[str]]]: (source)

Validate a group before it is stored.

Args:
  • e_uci: euci instance
  • name: name of the group
  • applications: list of application names
  • application_categories: list of application category tags
  • protocols: list of protocol names
  • protocol_categories: list of protocol category tags
  • config_name: config name of the group being edited, excluded from the name uniqueness check
Returns:
a tuple with the sanitized name and the sanitized members, indexed by UCI list name
Raises:
  • ValidationError: if the name or any member is invalid, or if the group has no members
def __validate_appgroup_members(kind: str, values: list[str]) -> list[str]: (source)

Validate the members of one kind, stripping and deduplicating them.

Args:
Returns:
the sanitized list of members
Raises:
  • ValidationError: if a member is empty, holds a character that would break the expression, or is unknown to both the engine and the catalog
def __validate_appgroups(e_uci: EUci, config_names: list[str], match_all: bool = False) -> list[str]: (source)

Validate the groups a rule matches. At least one is required: a rule with a source and no group would be an IP-level rule, which the firewall does better, and a rule with neither would expand to an empty criteria.

match_all is the way out of that requirement, and the only one: a rule that matches every flow names no group by definition, so naming one at the same time is a contradiction, not a narrowing.

def __validate_rule_name(name: str) -> str: (source)

Undocumented

def __validate_source(values: list[str]) -> list[str]: (source)

Validate the source of a rule: addresses, networks and ranges, IPv4 and IPv6 alike. An empty source means the rule matches every host.

__APPGROUP_FORBIDDEN_CHARS: str = (source)

Undocumented

Value
'''\'"\\;()
\r\t'''
__APPGROUP_MEMBER_ERRORS: dict[str, str] = (source)

Undocumented

Value
{'app': 'invalid_application',
 'app_category': 'invalid_application_category',
 'proto': 'invalid_protocol',
 'proto_category': 'invalid_protocol_category'}