diff options
| -rw-r--r-- | data/templates/conntrack/sysctl.conf.j2 | 3 | ||||
| -rw-r--r-- | debian/control | 1 | ||||
| -rw-r--r-- | interface-definitions/include/conntrack/log-common.xml.i | 20 | ||||
| -rw-r--r-- | interface-definitions/include/conntrack/log-protocols.xml.i | 26 | ||||
| -rw-r--r-- | interface-definitions/system_conntrack.xml.in | 81 | ||||
| -rwxr-xr-x | smoketest/scripts/cli/test_system_conntrack.py | 35 | ||||
| -rwxr-xr-x | src/conf_mode/system_conntrack.py | 21 | ||||
| -rwxr-xr-x | src/services/vyos-conntrack-logger | 458 | ||||
| -rw-r--r-- | src/systemd/vyos-conntrack-logger.service | 21 |
9 files changed, 620 insertions, 46 deletions
diff --git a/data/templates/conntrack/sysctl.conf.j2 b/data/templates/conntrack/sysctl.conf.j2 index 554512f4d..cd6c34ede 100644 --- a/data/templates/conntrack/sysctl.conf.j2 +++ b/data/templates/conntrack/sysctl.conf.j2 @@ -6,4 +6,5 @@ net.netfilter.nf_conntrack_max = {{ table_size }} net.ipv4.tcp_max_syn_backlog = {{ tcp.half_open_connections }} net.netfilter.nf_conntrack_tcp_loose = {{ '1' if tcp.loose is vyos_defined('enable') else '0' }} net.netfilter.nf_conntrack_tcp_max_retrans = {{ tcp.max_retrans }} -net.netfilter.nf_conntrack_acct = {{ '1' if flow_accounting is vyos_defined else '0' }}
\ No newline at end of file +net.netfilter.nf_conntrack_acct = {{ '1' if flow_accounting is vyos_defined else '0' }} +net.netfilter.nf_conntrack_timestamp = {{ '1' if log.timestamp is vyos_defined else '0' }}
\ No newline at end of file diff --git a/debian/control b/debian/control index 189a959b0..c5497b7dd 100644 --- a/debian/control +++ b/debian/control @@ -70,6 +70,7 @@ Depends: python3-netifaces, python3-paramiko, python3-passlib, + python3-pyroute2, python3-psutil, python3-pyhumps, python3-pystache, diff --git a/interface-definitions/include/conntrack/log-common.xml.i b/interface-definitions/include/conntrack/log-common.xml.i deleted file mode 100644 index 38799f8f4..000000000 --- a/interface-definitions/include/conntrack/log-common.xml.i +++ /dev/null @@ -1,20 +0,0 @@ -<!-- include start from conntrack/log-common.xml.i --> -<leafNode name="destroy"> - <properties> - <help>Log connection deletion</help> - <valueless/> - </properties> -</leafNode> -<leafNode name="new"> - <properties> - <help>Log connection creation</help> - <valueless/> - </properties> -</leafNode> -<leafNode name="update"> - <properties> - <help>Log connection updates</help> - <valueless/> - </properties> -</leafNode> -<!-- include end --> diff --git a/interface-definitions/include/conntrack/log-protocols.xml.i b/interface-definitions/include/conntrack/log-protocols.xml.i new file mode 100644 index 000000000..019250760 --- /dev/null +++ b/interface-definitions/include/conntrack/log-protocols.xml.i @@ -0,0 +1,26 @@ +<!-- include start from conntrack/log-protocols.xml.i --> +<leafNode name="icmp"> + <properties> + <help>Log connection tracking events for ICMP</help> + <valueless/> + </properties> +</leafNode> +<leafNode name="other"> + <properties> + <help>Log connection tracking events for all protocols other than TCP, UDP and ICMP</help> + <valueless/> + </properties> +</leafNode> +<leafNode name="tcp"> + <properties> + <help>Log connection tracking events for TCP</help> + <valueless/> + </properties> +</leafNode> +<leafNode name="udp"> + <properties> + <help>Log connection tracking events for UDP</help> + <valueless/> + </properties> +</leafNode> +<!-- include end --> diff --git a/interface-definitions/system_conntrack.xml.in b/interface-definitions/system_conntrack.xml.in index 0dfa2ea81..cd59d1308 100644 --- a/interface-definitions/system_conntrack.xml.in +++ b/interface-definitions/system_conntrack.xml.in @@ -223,41 +223,78 @@ </node> <node name="log"> <properties> - <help>Log connection tracking events per protocol</help> + <help>Log connection tracking</help> </properties> <children> - <node name="icmp"> + <node name="event"> <properties> - <help>Log connection tracking events for ICMP</help> + <help>Event type and protocol</help> </properties> <children> - #include <include/conntrack/log-common.xml.i> + <node name="destroy"> + <properties> + <help>Log connection deletion</help> + </properties> + <children> + #include <include/conntrack/log-protocols.xml.i> + </children> + </node> + <node name="new"> + <properties> + <help>Log connection creation</help> + </properties> + <children> + #include <include/conntrack/log-protocols.xml.i> + </children> + </node> + <node name="update"> + <properties> + <help>Log connection updates</help> + </properties> + <children> + #include <include/conntrack/log-protocols.xml.i> + </children> + </node> </children> </node> - <node name="other"> + <leafNode name="timestamp"> <properties> - <help>Log connection tracking events for all protocols other than TCP, UDP and ICMP</help> + <help>Log connection tracking events include flow-based timestamp</help> + <valueless/> </properties> - <children> - #include <include/conntrack/log-common.xml.i> - </children> - </node> - <node name="tcp"> + </leafNode> + <leafNode name="queue-size"> <properties> - <help>Log connection tracking events for TCP</help> + <help>Internal message queue size</help> + <valueHelp> + <format>u32:100-999999</format> + <description>Queue size</description> + </valueHelp> + <constraint> + <validator name="numeric" argument="--range 1-999999"/> + </constraint> + <constraintErrorMessage>Queue size must be between 100 and 999999</constraintErrorMessage> </properties> - <children> - #include <include/conntrack/log-common.xml.i> - </children> - </node> - <node name="udp"> + </leafNode> + <leafNode name="log-level"> <properties> - <help>Log connection tracking events for UDP</help> + <help>Set log-level. Log must be enable.</help> + <completionHelp> + <list>info debug</list> + </completionHelp> + <valueHelp> + <format>info</format> + <description>Info log level</description> + </valueHelp> + <valueHelp> + <format>debug</format> + <description>Debug log level</description> + </valueHelp> + <constraint> + <regex>(info|debug)</regex> + </constraint> </properties> - <children> - #include <include/conntrack/log-common.xml.i> - </children> - </node> + </leafNode> </children> </node> <node name="modules"> diff --git a/smoketest/scripts/cli/test_system_conntrack.py b/smoketest/scripts/cli/test_system_conntrack.py index 3ae7b6217..c07fdce77 100755 --- a/smoketest/scripts/cli/test_system_conntrack.py +++ b/smoketest/scripts/cli/test_system_conntrack.py @@ -20,7 +20,7 @@ import unittest from base_vyostest_shim import VyOSUnitTestSHIM from vyos.firewall import find_nftables_rule -from vyos.utils.file import read_file +from vyos.utils.file import read_file, read_json base_path = ['system', 'conntrack'] @@ -28,6 +28,9 @@ def get_sysctl(parameter): tmp = parameter.replace(r'.', r'/') return read_file(f'/proc/sys/{tmp}') +def get_logger_config(): + return read_json('/run/vyos-conntrack-logger.conf') + class TestSystemConntrack(VyOSUnitTestSHIM.TestCase): @classmethod def setUpClass(cls): @@ -280,5 +283,35 @@ class TestSystemConntrack(VyOSUnitTestSHIM.TestCase): self.verify_nftables(nftables6_search, 'ip6 vyos_conntrack') self.cli_delete(['firewall']) + + def test_conntrack_log(self): + expected_config = { + 'event': { + 'destroy': {}, + 'new': {}, + 'update': {}, + }, + 'queue_size': '10000' + } + self.cli_set(base_path + ['log', 'event', 'destroy']) + self.cli_set(base_path + ['log', 'event', 'new']) + self.cli_set(base_path + ['log', 'event', 'update']) + self.cli_set(base_path + ['log', 'queue-size', '10000']) + self.cli_commit() + self.assertEqual(expected_config, get_logger_config()) + self.assertEqual('0', get_sysctl('net.netfilter.nf_conntrack_timestamp')) + + for event in ['destroy', 'new', 'update']: + for proto in ['icmp', 'other', 'tcp', 'udp']: + self.cli_set(base_path + ['log', 'event', event, proto]) + expected_config['event'][event][proto] = {} + self.cli_set(base_path + ['log', 'timestamp']) + expected_config['timestamp'] = {} + self.cli_commit() + + self.assertEqual(expected_config, get_logger_config()) + self.assertEqual('1', get_sysctl('net.netfilter.nf_conntrack_timestamp')) + + if __name__ == '__main__': unittest.main(verbosity=2) diff --git a/src/conf_mode/system_conntrack.py b/src/conf_mode/system_conntrack.py index aa290788c..2529445bf 100755 --- a/src/conf_mode/system_conntrack.py +++ b/src/conf_mode/system_conntrack.py @@ -13,7 +13,7 @@ # # You should have received a copy of the GNU General Public License # along with this program. If not, see <http://www.gnu.org/licenses/>. - +import json import os from sys import exit @@ -24,7 +24,8 @@ from vyos.configdep import set_dependents, call_dependents from vyos.utils.dict import dict_search from vyos.utils.dict import dict_search_args from vyos.utils.dict import dict_search_recursive -from vyos.utils.process import cmd +from vyos.utils.file import write_file +from vyos.utils.process import cmd, call from vyos.utils.process import rc_cmd from vyos.template import render from vyos import ConfigError @@ -34,6 +35,7 @@ airbag.enable() conntrack_config = r'/etc/modprobe.d/vyatta_nf_conntrack.conf' sysctl_file = r'/run/sysctl/10-vyos-conntrack.conf' nftables_ct_file = r'/run/nftables-ct.conf' +vyos_conntrack_logger_config = r'/run/vyos-conntrack-logger.conf' # Every ALG (Application Layer Gateway) consists of either a Kernel Object # also called a Kernel Module/Driver or some rules present in iptables @@ -113,6 +115,7 @@ def get_config(config=None): return conntrack + def verify(conntrack): for inet in ['ipv4', 'ipv6']: if dict_search_args(conntrack, 'ignore', inet, 'rule') != None: @@ -181,6 +184,11 @@ def generate(conntrack): if not os.path.exists(nftables_ct_file): conntrack['first_install'] = True + if 'log' not in conntrack: + # Remove old conntrack-logger config and return + if os.path.exists(vyos_conntrack_logger_config): + os.unlink(vyos_conntrack_logger_config) + # Determine if conntrack is needed conntrack['ipv4_firewall_action'] = 'return' conntrack['ipv6_firewall_action'] = 'return' @@ -199,6 +207,11 @@ def generate(conntrack): render(conntrack_config, 'conntrack/vyos_nf_conntrack.conf.j2', conntrack) render(sysctl_file, 'conntrack/sysctl.conf.j2', conntrack) render(nftables_ct_file, 'conntrack/nftables-ct.j2', conntrack) + + if 'log' in conntrack: + log_conf_json = json.dumps(conntrack['log'], indent=4) + write_file(vyos_conntrack_logger_config, log_conf_json) + return None def apply(conntrack): @@ -243,8 +256,12 @@ def apply(conntrack): # See: https://bugzilla.redhat.com/show_bug.cgi?id=1264080 cmd(f'sysctl -f {sysctl_file}') + if 'log' in conntrack: + call(f'systemctl restart vyos-conntrack-logger.service') + return None + if __name__ == '__main__': try: c = get_config() diff --git a/src/services/vyos-conntrack-logger b/src/services/vyos-conntrack-logger new file mode 100755 index 000000000..9c31b465f --- /dev/null +++ b/src/services/vyos-conntrack-logger @@ -0,0 +1,458 @@ +#!/usr/bin/env python3 +# +# Copyright (C) 2024 VyOS maintainers and contributors +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License version 2 or later as +# published by the Free Software Foundation. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see <http://www.gnu.org/licenses/>. + +import argparse +import grp +import logging +import multiprocessing +import os +import queue +import signal +import socket +import threading +from datetime import timedelta +from pathlib import Path +from time import sleep +from typing import Dict, AnyStr + +from pyroute2 import conntrack +from pyroute2.netlink import nfnetlink +from pyroute2.netlink.nfnetlink import NFNL_SUBSYS_CTNETLINK +from pyroute2.netlink.nfnetlink.nfctsocket import nfct_msg, \ + IPCTNL_MSG_CT_DELETE, IPCTNL_MSG_CT_NEW, IPS_SEEN_REPLY, \ + IPS_OFFLOAD, IPS_ASSURED + +from vyos.utils.file import read_json + + +shutdown_event = multiprocessing.Event() + +logging.basicConfig(level=logging.INFO, format='%(message)s') +logger = logging.getLogger(__name__) + + +class DebugFormatter(logging.Formatter): + def format(self, record): + self._style._fmt = '[%(asctime)s] %(levelname)s: %(message)s' + return super().format(record) + + +def set_log_level(level: str) -> None: + if level == 'debug': + logger.setLevel(logging.DEBUG) + logger.parent.handlers[0].setFormatter(DebugFormatter()) + else: + logger.setLevel(logging.INFO) + + +EVENT_NAME_TO_GROUP = { + 'new': nfnetlink.NFNLGRP_CONNTRACK_NEW, + 'update': nfnetlink.NFNLGRP_CONNTRACK_UPDATE, + 'destroy': nfnetlink.NFNLGRP_CONNTRACK_DESTROY +} + +# https://github.com/torvalds/linux/blob/1dfe225e9af5bd3399a1dbc6a4df6a6041ff9c23/include/uapi/linux/netfilter/nf_conntrack_tcp.h#L9 +TCP_CONNTRACK_SYN_SENT = 1 +TCP_CONNTRACK_SYN_RECV = 2 +TCP_CONNTRACK_ESTABLISHED = 3 +TCP_CONNTRACK_FIN_WAIT = 4 +TCP_CONNTRACK_CLOSE_WAIT = 5 +TCP_CONNTRACK_LAST_ACK = 6 +TCP_CONNTRACK_TIME_WAIT = 7 +TCP_CONNTRACK_CLOSE = 8 +TCP_CONNTRACK_LISTEN = 9 +TCP_CONNTRACK_MAX = 10 +TCP_CONNTRACK_IGNORE = 11 +TCP_CONNTRACK_RETRANS = 12 +TCP_CONNTRACK_UNACK = 13 +TCP_CONNTRACK_TIMEOUT_MAX = 14 + +TCP_CONNTRACK_TO_NAME = { + TCP_CONNTRACK_SYN_SENT: "SYN_SENT", + TCP_CONNTRACK_SYN_RECV: "SYN_RECV", + TCP_CONNTRACK_ESTABLISHED: "ESTABLISHED", + TCP_CONNTRACK_FIN_WAIT: "FIN_WAIT", + TCP_CONNTRACK_CLOSE_WAIT: "CLOSE_WAIT", + TCP_CONNTRACK_LAST_ACK: "LAST_ACK", + TCP_CONNTRACK_TIME_WAIT: "TIME_WAIT", + TCP_CONNTRACK_CLOSE: "CLOSE", + TCP_CONNTRACK_LISTEN: "LISTEN", + TCP_CONNTRACK_MAX: "MAX", + TCP_CONNTRACK_IGNORE: "IGNORE", + TCP_CONNTRACK_RETRANS: "RETRANS", + TCP_CONNTRACK_UNACK: "UNACK", + TCP_CONNTRACK_TIMEOUT_MAX: "TIMEOUT_MAX", +} + +# https://github.com/torvalds/linux/blob/1dfe225e9af5bd3399a1dbc6a4df6a6041ff9c23/include/uapi/linux/netfilter/nf_conntrack_sctp.h#L8 +SCTP_CONNTRACK_CLOSED = 1 +SCTP_CONNTRACK_COOKIE_WAIT = 2 +SCTP_CONNTRACK_COOKIE_ECHOED = 3 +SCTP_CONNTRACK_ESTABLISHED = 4 +SCTP_CONNTRACK_SHUTDOWN_SENT = 5 +SCTP_CONNTRACK_SHUTDOWN_RECD = 6 +SCTP_CONNTRACK_SHUTDOWN_ACK_SENT = 7 +SCTP_CONNTRACK_HEARTBEAT_SENT = 8 +SCTP_CONNTRACK_HEARTBEAT_ACKED = 9 # no longer used +SCTP_CONNTRACK_MAX = 10 + +SCTP_CONNTRACK_TO_NAME = { + SCTP_CONNTRACK_CLOSED: 'CLOSED', + SCTP_CONNTRACK_COOKIE_WAIT: 'COOKIE_WAIT', + SCTP_CONNTRACK_COOKIE_ECHOED: 'COOKIE_ECHOED', + SCTP_CONNTRACK_ESTABLISHED: 'ESTABLISHED', + SCTP_CONNTRACK_SHUTDOWN_SENT: 'SHUTDOWN_SENT', + SCTP_CONNTRACK_SHUTDOWN_RECD: 'SHUTDOWN_RECD', + SCTP_CONNTRACK_SHUTDOWN_ACK_SENT: 'SHUTDOWN_ACK_SENT', + SCTP_CONNTRACK_HEARTBEAT_SENT: 'HEARTBEAT_SENT', + SCTP_CONNTRACK_HEARTBEAT_ACKED: 'HEARTBEAT_ACKED', + SCTP_CONNTRACK_MAX: 'MAX', +} + +PROTO_CONNTRACK_TO_NAME = { + 'TCP': TCP_CONNTRACK_TO_NAME, + 'SCTP': SCTP_CONNTRACK_TO_NAME +} + +SUPPORTED_PROTO_TO_NAME = { + socket.IPPROTO_ICMP: 'icmp', + socket.IPPROTO_TCP: 'tcp', + socket.IPPROTO_UDP: 'udp', +} + +PROTO_TO_NAME = { + socket.IPPROTO_ICMPV6: 'icmpv6', + socket.IPPROTO_SCTP: 'sctp', + socket.IPPROTO_GRE: 'gre', +} + +PROTO_TO_NAME.update(SUPPORTED_PROTO_TO_NAME) + + +def sig_handler(signum, frame): + process_name = multiprocessing.current_process().name + logger.debug(f'[{process_name}]: {"Shutdown" if signum == signal.SIGTERM else "Reload"} signal received...') + shutdown_event.set() + + +def format_flow_data(data: Dict) -> AnyStr: + """ + Formats the flow event data into a string suitable for logging. + """ + key_format = { + 'SRC_PORT': 'sport', + 'DST_PORT': 'dport' + } + message = f"src={data['ADDR'].get('SRC')} dst={data['ADDR'].get('DST')}" + + for key in ['SRC_PORT', 'DST_PORT', 'TYPE', 'CODE', 'ID']: + tmp = data['PROTO'].get(key) + if tmp is not None: + key = key_format.get(key, key) + message += f" {key.lower()}={tmp}" + + if 'COUNTERS' in data: + for key in ['PACKETS', 'BYTES']: + tmp = data['COUNTERS'].get(key) + if tmp is not None: + message += f" {key.lower()}={tmp}" + + return message + + +def format_event_message(event: Dict) -> AnyStr: + """ + Formats the internal parsed event data into a string suitable for logging. + """ + event_type = f"[{event['COMMON']['EVENT_TYPE'].upper()}]" + message = f"{event_type:<{9}} {event['COMMON']['ID']} " \ + f"{event['ORIG']['PROTO'].get('NAME'):<{8}} " \ + f"{event['ORIG']['PROTO'].get('NUMBER')} " + + tmp = event['COMMON']['TIME_OUT'] + if tmp is not None: message += f"{tmp} " + + if proto_info := event['COMMON'].get('PROTO_INFO'): + message += f"{proto_info.get('STATE_NAME')} " + + for key in ['ORIG', 'REPLY']: + message += f"{format_flow_data(event[key])} " + if key == 'ORIG' and not (event['COMMON']['STATUS'] & IPS_SEEN_REPLY): + message += f"[UNREPLIED] " + + tmp = event['COMMON']['MARK'] + if tmp is not None: message += f"mark={tmp} " + + if event['COMMON']['STATUS'] & IPS_OFFLOAD: message += f" [OFFLOAD] " + elif event['COMMON']['STATUS'] & IPS_ASSURED: message += f" [ASSURED] " + + if tmp := event['COMMON']['PORTID']: message += f"portid={tmp} " + if tstamp := event['COMMON'].get('TIMESTAMP'): + message += f"start={tstamp['START']} stop={tstamp['STOP']} " + delta_ns = tstamp['STOP'] - tstamp['START'] + delta_s = delta_ns // 1e9 + remaining_ns = delta_ns % 1e9 + delta = timedelta(seconds=delta_s, microseconds=remaining_ns / 1000) + message += f"delta={delta.total_seconds()} " + + return message + + +def parse_event_type(header: Dict) -> AnyStr: + """ + Extract event type from nfct_msg. new, update, destroy + """ + event_type = 'unknown' + if header['type'] == IPCTNL_MSG_CT_DELETE | (NFNL_SUBSYS_CTNETLINK << 8): + event_type = 'destroy' + elif header['type'] == IPCTNL_MSG_CT_NEW | (NFNL_SUBSYS_CTNETLINK << 8): + event_type = 'update' + if header['flags']: + event_type = 'new' + return event_type + + +def parse_proto(cta: nfct_msg.cta_tuple) -> Dict: + """ + Extract proto info from nfct_msg. src/dst port, code, type, id + """ + data = dict() + + cta_proto = cta.get_attr('CTA_TUPLE_PROTO') + proto_num = cta_proto.get_attr('CTA_PROTO_NUM') + + data['NUMBER'] = proto_num + data['NAME'] = PROTO_TO_NAME.get(proto_num, 'unknown') + + if proto_num in (socket.IPPROTO_ICMP, socket.IPPROTO_ICMPV6): + pref = 'CTA_PROTO_ICMP' + if proto_num == socket.IPPROTO_ICMPV6: pref += 'V6' + keys = ['TYPE', 'CODE', 'ID'] + else: + pref = 'CTA_PROTO' + keys = ['SRC_PORT', 'DST_PORT'] + + for key in keys: + data[key] = cta_proto.get_attr(f'{pref}_{key}') + + return data + + +def parse_proto_info(cta: nfct_msg.cta_protoinfo) -> Dict: + """ + Extract proto state and state name from nfct_msg + """ + data = dict() + if not cta: + return data + + for proto in ['TCP', 'SCTP']: + if proto_info := cta.get_attr(f'CTA_PROTOINFO_{proto}'): + data['STATE'] = proto_info.get_attr(f'CTA_PROTOINFO_{proto}_STATE') + data['STATE_NAME'] = PROTO_CONNTRACK_TO_NAME.get(proto, {}).get(data['STATE'], 'unknown') + return data + + +def parse_timestamp(cta: nfct_msg.cta_timestamp) -> Dict: + """ + Extract timestamp from nfct_msg + """ + data = dict() + if not cta: + return data + data['START'] = cta.get_attr('CTA_TIMESTAMP_START') + data['STOP'] = cta.get_attr('CTA_TIMESTAMP_STOP') + + return data + + +def parse_ip_addr(family: int, cta: nfct_msg.cta_tuple) -> Dict: + """ + Extract ip adr from nfct_msg + """ + data = dict() + cta_ip = cta.get_attr('CTA_TUPLE_IP') + + if family == socket.AF_INET: + pref = 'CTA_IP_V4' + elif family == socket.AF_INET6: + pref = 'CTA_IP_V6' + else: + logger.error(f'Undefined INET: {family}') + raise NotImplementedError(family) + + for direct in ['SRC', 'DST']: + data[direct] = cta_ip.get_attr(f'{pref}_{direct}') + + return data + + +def parse_counters(cta: nfct_msg.cta_counters) -> Dict: + """ + Extract counters from nfct_msg + """ + data = dict() + if not cta: + return data + + for key in ['PACKETS', 'BYTES']: + tmp = cta.get_attr(f'CTA_COUNTERS_{key}') + if tmp is None: + tmp = cta.get_attr(f'CTA_COUNTERS32_{key}') + data['key'] = tmp + + return data + + +def is_need_to_log(event_type: AnyStr, proto_num: int, conf_event: Dict): + """ + Filter message by event type and protocols + """ + conf = conf_event.get(event_type) + if conf == {} or conf.get(SUPPORTED_PROTO_TO_NAME.get(proto_num, 'other')) is not None: + return True + return False + + +def parse_conntrack_event(msg: nfct_msg, conf_event: Dict) -> Dict: + """ + Convert nfct_msg to internal data dict. + """ + data = dict() + event_type = parse_event_type(msg['header']) + proto_num = msg.get_nested('CTA_TUPLE_ORIG', 'CTA_TUPLE_PROTO', 'CTA_PROTO_NUM') + + if not is_need_to_log(event_type, proto_num, conf_event): + return data + + data = { + 'COMMON': { + 'ID': msg.get_attr('CTA_ID'), + 'EVENT_TYPE': event_type, + 'TIME_OUT': msg.get_attr('CTA_TIMEOUT'), + 'MARK': msg.get_attr('CTA_MARK'), + 'PORTID': msg['header'].get('pid'), + 'PROTO_INFO': parse_proto_info(msg.get_attr('CTA_PROTOINFO')), + 'STATUS': msg.get_attr('CTA_STATUS'), + 'TIMESTAMP': parse_timestamp(msg.get_attr('CTA_TIMESTAMP')) + }, + 'ORIG': {}, + 'REPLY': {}, + } + + for direct in ['ORIG', 'REPLY']: + data[direct]['ADDR'] = parse_ip_addr(msg['nfgen_family'], msg.get_attr(f'CTA_TUPLE_{direct}')) + data[direct]['PROTO'] = parse_proto(msg.get_attr(f'CTA_TUPLE_{direct}')) + data[direct]['COUNTERS'] = parse_counters(msg.get_attr(f'CTA_COUNTERS_{direct}')) + + return data + + +def worker(ct: conntrack.Conntrack, shutdown_event: multiprocessing.Event, conf_event: Dict): + """ + Main function of parser worker process + """ + process_name = multiprocessing.current_process().name + logger.debug(f'[{process_name}] started') + timeout = 0.1 + while not shutdown_event.is_set(): + if not ct.buffer_queue.empty(): + try: + for msg in ct.get(): + parsed_event = parse_conntrack_event(msg, conf_event) + if parsed_event: + message = format_event_message(parsed_event) + if logger.level == logging.DEBUG: + logger.debug(f"[{process_name}]: {message} raw: {msg}") + else: + logger.info(message) + except queue.Full: + logger.error("Conntrack message queue if full.") + except Exception as e: + logger.error(f"Error in queue: {e.__class__} {e}") + else: + sleep(timeout) + + +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument('-c', + '--config', + action='store', + help='Path to vyos-conntrack-logger configuration', + required=True, + type=Path) + + args = parser.parse_args() + try: + config = read_json(args.config) + except Exception as err: + logger.error(f'Configuration file "{args.config}" does not exist or malformed: {err}') + exit(1) + + set_log_level(config.get('log_level', 'info')) + + signal.signal(signal.SIGHUP, sig_handler) + signal.signal(signal.SIGTERM, sig_handler) + + if 'event' in config: + event_groups = list(config.get('event').keys()) + else: + logger.error(f'Configuration is wrong. Event filter is empty.') + exit(1) + + conf_event = config['event'] + qsize = config.get('queue_size') + ct = conntrack.Conntrack(async_qsize=int(qsize) if qsize else None) + ct.buffer_queue = multiprocessing.Queue(ct.async_qsize) + ct.bind(async_cache=True) + + for name in event_groups: + if group := EVENT_NAME_TO_GROUP.get(name): + ct.add_membership(group) + else: + logger.error(f'Unexpected event group {name}') + processes = list() + try: + for _ in range(multiprocessing.cpu_count()): + p = multiprocessing.Process(target=worker, args=(ct, + shutdown_event, + conf_event)) + processes.append(p) + p.start() + logger.info('Conntrack socket bound and listening for messages.') + + while not shutdown_event.is_set(): + if not ct.pthread.is_alive(): + if ct.buffer_queue.qsize()/ct.async_qsize < 0.9: + if not shutdown_event.is_set(): + logger.debug('Restart listener thread') + # restart listener thread after queue overloaded when queue size low than 90% + ct.pthread = threading.Thread( + name="Netlink async cache", target=ct.async_recv + ) + ct.pthread.daemon = True + ct.pthread.start() + else: + sleep(0.1) + finally: + for p in processes: + p.join() + if not p.is_alive(): + logger.debug(f"[{p.name}]: finished") + ct.close() + logging.info("Conntrack socket closed.") + exit() diff --git a/src/systemd/vyos-conntrack-logger.service b/src/systemd/vyos-conntrack-logger.service new file mode 100644 index 000000000..9bc1d857b --- /dev/null +++ b/src/systemd/vyos-conntrack-logger.service @@ -0,0 +1,21 @@ +[Unit] +Description=VyOS conntrack logger daemon + +# Seemingly sensible way to say "as early as the system is ready" +# All vyos-configd needs is read/write mounted root +After=conntrackd.service + +[Service] +ExecStart=/usr/bin/python3 -u /usr/libexec/vyos/services/vyos-conntrack-logger -c /run/vyos-conntrack-logger.conf +Type=idle + +SyslogIdentifier=vyos-conntrack-logger +SyslogFacility=daemon + +Restart=on-failure + +User=root +Group=vyattacfg + +[Install] +WantedBy=multi-user.target |