diff --git a/node_cli/core/nftables.py b/node_cli/core/nftables.py index e0059fff..01efb336 100644 --- a/node_cli/core/nftables.py +++ b/node_cli/core/nftables.py @@ -27,14 +27,16 @@ from typing import Optional from node_cli.configs import ( + DEFAULT_NODE_BASE_PORT, ENV, NFTABLES_CHAIN_CONFIG_WILDCARD, NFTABLES_CHAIN_FOLDER_PATH, NFTABLES_MAIN_CONFIG_PATH, NFTABLES_SKALE_BASE_CONFIG_PATH, NFTABLES_USER_CONFIG_PATH, + NODE_CONFIG_PATH, ) -from node_cli.utils.helper import get_ssh_port, run_cmd +from node_cli.utils.helper import get_ssh_port, read_json, run_cmd logger = logging.getLogger(__name__) @@ -66,6 +68,27 @@ class SGXPort: CHAIN_PRIORITY = 1 HOOK = 'input' POLICY = 'accept' +POLICY_DROP = 'drop' + +DYNAMIC_CHAIN_PREFIX = 'skale-' + +# sChain base ports are allocated as node_base_port + schain_index * 64 +# (PORTS_PER_SCHAIN in skale.py); 128 slots cover every possible allocation +SCHAIN_PORTS_PER_NODE = 128 * 64 +SCHAIN_BASE_PORT_ENV = 'SCHAIN_BASE_PORT' +FIREWALL_DEFAULT_DROP_ENV = 'FIREWALL_DEFAULT_DROP' +MIN_SCHAIN_BASE_PORT = 2000 +MAX_PORT = 65535 + +ICMPV6_ACCEPT_TYPES = ( + 'destination-unreachable', + 'packet-too-big', + 'time-exceeded', + 'parameter-problem', + 'nd-router-advert', + 'nd-neighbor-solicit', + 'nd-neighbor-advert', +) try: @@ -180,24 +203,33 @@ def update_chain_policy( family = family or self.family table = table or self.table if self.chain_exists(chain, family=family): - cmd = [ - 'nft', - 'add', - 'chain', - family, - table, - chain, - '{', - 'policy', - POLICY, - ';', - '}', - ] - run_cmd(cmd) + cmd = f'add chain {family} {table} {chain} {{ policy {policy} ; }}' + rc, output, error = self.nft.cmd(cmd) + if rc != 0: + raise NFTablesError(f'Failed to set policy {policy} on {chain}: {error}') logger.info('Updated chain policy: %s %s', chain, policy) else: logger.info('Chain %s does not exist', chain) + def get_chain_policy(self, chain: str) -> Optional[str]: + """Return the policy of a chain in the managed table or None.""" + try: + rc, output, error = self.nft.cmd(f'list chain {self.family} {self.table} {chain}') + if rc != 0: + return None + data = json.loads(output) + if not isinstance(data, dict): + return None + for item in data.get('nftables', []): + if not isinstance(item, dict): + continue + chain_data = item.get('chain') + if isinstance(chain_data, dict) and chain_data.get('name') == chain: + return chain_data.get('policy') + except (TypeError, ValueError) as e: + logger.error('Failed to get policy of chain %s: %s', chain, e) + return None + def table_exists(self) -> bool: try: rc, output, error = self.nft.cmd(f'list table {self.family} {self.table}') @@ -339,7 +371,7 @@ def add_rule(self, rule: Rule) -> None: { 'match': { 'op': '==', - 'left': {'payload': {'protocol': 'tcp', 'field': 'dport'}}, + 'left': {'payload': {'protocol': rule.protocol, 'field': 'dport'}}, 'right': rule.first_port, } } @@ -349,16 +381,16 @@ def add_rule(self, rule: Rule) -> None: { 'match': { 'op': '==', - 'left': {'payload': {'protocol': 'tcp', 'field': 'dport'}}, + 'left': {'payload': {'protocol': rule.protocol, 'field': 'dport'}}, 'right': {'range': [rule.first_port, rule.last_port]}, } } ) - elif rule.protocol == 'icmp' and rule.icmp_type: + elif rule.protocol in ['icmp', 'icmpv6'] and rule.icmp_type: expr.append( { 'match': { - 'left': {'payload': {'protocol': 'icmp', 'field': 'type'}}, + 'left': {'payload': {'protocol': rule.protocol, 'field': 'type'}}, 'op': '==', 'right': rule.icmp_type, } @@ -410,7 +442,7 @@ def remove_rule(self, rule: Rule) -> None: { 'match': { 'op': '==', - 'left': {'payload': {'protocol': 'tcp', 'field': 'dport'}}, + 'left': {'payload': {'protocol': rule.protocol, 'field': 'dport'}}, 'right': rule.first_port, } } @@ -420,7 +452,7 @@ def remove_rule(self, rule: Rule) -> None: { 'match': { 'op': '==', - 'left': {'payload': {'protocol': 'tcp', 'field': 'dport'}}, + 'left': {'payload': {'protocol': rule.protocol, 'field': 'dport'}}, 'right': {'range': [rule.first_port, rule.last_port]}, } } @@ -527,6 +559,225 @@ def add_loopback_rule(self, chain) -> None: else: logger.info('Loopback rule already exists in chain %s', chain) + def get_dynamic_chain_port_ranges(self) -> list[tuple[str, int, int]]: + """Min/max tcp dport covered by each dynamic skale-admin chain.""" + try: + rc, output, error = self.nft.cmd(f'list table {self.family} {self.table}') + if rc != 0: + if error and 'No such file or directory' in error: + return [] + raise NFTablesError(f'Failed to list table {self.table}: {error}') + data = json.loads(output) + except NFTablesError: + raise + except Exception as e: + logger.error('Failed to get dynamic chain ranges: %s', e) + raise NFTablesError(e) + + ports: dict[str, list[int]] = {} + for item in data.get('nftables', []): + rule = item.get('rule') + if not rule or not rule.get('chain', '').startswith(DYNAMIC_CHAIN_PREFIX): + continue + for statement in rule.get('expr', []): + match = statement.get('match', {}) + if match.get('left', {}).get('payload', {}).get('field') != 'dport': + continue + right = match.get('right') + chain_ports = ports.setdefault(rule['chain'], []) + if isinstance(right, dict) and 'range' in right: + chain_ports.extend(right['range']) + elif isinstance(right, int): + chain_ports.append(right) + return [(chain, min(values), max(values)) for chain, values in ports.items() if values] + + def validate_dynamic_ranges(self, envelope: tuple[int, int]) -> None: + """Ensure ports of every dynamic skale-admin chain fit into the envelope.""" + for chain, first_port, last_port in self.get_dynamic_chain_port_ranges(): + if first_port < envelope[0] or last_port > envelope[1]: + raise NFTablesError( + f'Ports {first_port}-{last_port} of dynamic chain {chain} are outside ' + f'of the allowed sChain ports range {envelope[0]}-{envelope[1]}. ' + f'Set {SCHAIN_BASE_PORT_ENV} env variable to the base port the node ' + 'was registered with and rerun the command' + ) + + def verify_critical_accepts(self) -> None: + """Ensure lockout-critical accept rules are in place before setting drop policy.""" + conntrack_expr = [ + { + 'match': { + 'left': {'ct': {'key': 'state'}}, + 'op': 'in', + 'right': ['established', 'related'], + } + }, + {'counter': None}, + {'accept': None}, + ] + ssh_expr = [ + { + 'match': { + 'op': '==', + 'left': {'payload': {'protocol': 'tcp', 'field': 'dport'}}, + 'right': get_ssh_port(), + } + }, + {'counter': None}, + {'accept': None}, + ] + for name, expr in (('conntrack', conntrack_expr), ('ssh', ssh_expr)): + if not self.rule_exists(self.chain, expr): + raise NFTablesError( + f'Refusing to set drop policy: {name} accept rule is missing ' + f'in chain {self.chain}' + ) + + def ensure_default_drop(self, envelope: tuple[int, int]) -> None: + """Switch the skale chain policy to drop after validating the accepts.""" + self.validate_dynamic_ranges(envelope) + self.verify_critical_accepts() + if self.get_chain_policy(self.chain) != POLICY_DROP: + self.update_chain_policy(chain=self.chain, policy=POLICY_DROP) + + def ensure_default_accept(self) -> None: + """Rollback path: switch the skale chain policy back to accept. + + Flips whenever the policy cannot be confirmed as accept, so an + unreadable policy does not silently skip the rollback. + """ + if self.get_chain_policy(self.chain) != POLICY: + self.update_chain_policy(chain=self.chain, policy=POLICY) + + def delete_rule_by_handle(self, handle: int) -> None: + cmd = { + 'nftables': [ + { + 'delete': { + 'rule': { + 'family': self.family, + 'table': self.table, + 'chain': self.chain, + 'handle': handle, + } + } + } + ] + } + self.execute_cmd(cmd) + + def remove_stale_envelope_rules(self, envelope: tuple[int, int]) -> None: + """Remove sChain envelope accepts anchored at a different base port.""" + for rule in self.get_rules(self.chain): + expr = rule.get('expr', []) + if {'accept': None} not in expr: + continue + for statement in expr: + match = statement.get('match', {}) + right = match.get('right') + if ( + match.get('left', {}).get('payload', {}).get('field') == 'dport' + and isinstance(right, dict) + and 'range' in right + and right['range'][1] - right['range'][0] == SCHAIN_PORTS_PER_NODE - 1 + and tuple(right['range']) != envelope + and rule.get('handle') is not None + ): + logger.info('Removing stale envelope rule %s', right['range']) + self.delete_rule_by_handle(rule['handle']) + + @staticmethod + def _normalized_expr(expr: list[dict]) -> list[dict]: + return [{'counter': None} if 'counter' in statement else statement for statement in expr] + + def remove_misordered_udp_drop(self) -> None: + """Delete the blanket udp drop when it shadows the udp DNS accept. + + Rulesets created before the udp payload fix have the drop above the + accept; setup re-adds the drop after all accept rules. + """ + udp_drop = [ + { + 'match': { + 'left': {'payload': {'protocol': 'ip', 'field': 'protocol'}}, + 'op': '==', + 'right': 'udp', + } + }, + {'counter': None}, + {'drop': None}, + ] + udp_dns_accept = [ + { + 'match': { + 'op': '==', + 'left': {'payload': {'protocol': 'udp', 'field': 'dport'}}, + 'right': ServicePort.DNS, + } + }, + {'counter': None}, + {'accept': None}, + ] + drop_handle, drop_index, accept_index = None, None, None + for index, rule in enumerate(self.get_rules(self.chain)): + expr = self._normalized_expr(rule.get('expr', [])) + if expr == udp_drop: + drop_handle, drop_index = rule.get('handle'), index + elif expr == udp_dns_accept: + accept_index = index + if drop_handle is not None and (accept_index is None or drop_index < accept_index): + logger.info('Removing misordered udp drop rule') + self.delete_rule_by_handle(drop_handle) + + def remove_source_quench_rule(self) -> None: + """Remove the legacy icmp source-quench accept (deprecated by RFC 6633).""" + expr = [ + { + 'match': { + 'left': {'payload': {'protocol': 'icmp', 'field': 'type'}}, + 'op': '==', + 'right': 'source-quench', + } + }, + {'counter': None}, + {'accept': None}, + ] + for rule in self.get_rules(self.chain): + if ( + self._normalized_expr(rule.get('expr', [])) == expr + and rule.get('handle') is not None + ): + logger.info('Removing legacy source-quench rule') + self.delete_rule_by_handle(rule['handle']) + + def apply_user_rules(self) -> None: + """Load user.conf rules into the live chain. + + The file is included into the saved config, but the live chain is + managed through the API - without this, rules added to the file would + apply only after a reboot and would be missing from the live chain + when the policy flips to drop. + """ + if not os.path.isfile(NFTABLES_USER_CONFIG_PATH): + return + with open(NFTABLES_USER_CONFIG_PATH) as user_config: + lines = [line.strip() for line in user_config.readlines()] + lines = [line for line in lines if line and not line.startswith('#')] + if not lines: + return + current_rules = self.get_base_ruleset() + # insert in reverse to keep the file order at the top of the chain, + # mirroring the include position in the saved config + for line in reversed(lines): + if line in current_rules: + continue + rc, output, error = self.nft.cmd( + f'insert rule {self.family} {self.table} {self.chain} {line}' + ) + if rc != 0: + raise NFTablesError(f'Failed to apply user.conf rule "{line}": {error}') + logger.info('Applied user.conf rule: %s', line) + def get_base_ruleset(self) -> str: self.nft.set_json_output(False) try: @@ -538,13 +789,24 @@ def get_base_ruleset(self) -> str: finally: self.nft.set_json_output(True) + def setup_firewall( + self, enable_monitoring: bool = False, keep_accept_policy: bool = False + ) -> None: + """Setup firewall rules. - def setup_firewall(self, enable_monitoring: bool = False) -> None: - """Setup firewall rules.""" + keep_accept_policy leaves the chain on the accept policy for this + run - used when the envelope base port is not known yet (fresh + passive init, where skale-admin computes it only after the + containers start). + """ logger.info('Configuring firewall rules') + envelope = get_schain_ports_envelope() + default_drop = firewall_default_drop_enabled() and not keep_accept_policy try: self.create_table_if_not_exists() + if default_drop: + self.validate_dynamic_ranges(envelope) base_chains_config = {'skale': {'hook': 'input', 'policy': 'accept'}} @@ -568,13 +830,30 @@ def setup_firewall(self, enable_monitoring: bool = False) -> None: for port in tcp_ports: self.add_rule(Rule(chain=self.chain, protocol='tcp', first_port=port)) + self.remove_misordered_udp_drop() self.add_rule(Rule(chain=self.chain, protocol='udp', first_port=ServicePort.DNS)) self.add_loopback_rule(chain=self.chain) - icmp_types = ['destination-unreachable', 'source-quench', 'time-exceeded'] + self.remove_source_quench_rule() + icmp_types = ['destination-unreachable', 'time-exceeded'] for icmp_type in icmp_types: self.add_rule(Rule(chain=self.chain, protocol='icmp', icmp_type=icmp_type)) + for icmpv6_type in ICMPV6_ACCEPT_TYPES: + self.add_rule(Rule(chain=self.chain, protocol='icmpv6', icmp_type=icmpv6_type)) + + # Fine-grained filtering inside the envelope is enforced by the + # dynamic skale-admin chains that run earlier (priority 0) + self.remove_stale_envelope_rules(envelope) + self.add_rule( + Rule( + chain=self.chain, + protocol='tcp', + first_port=envelope[0], + last_port=envelope[1], + ) + ) + self.add_drop_rule( Rule( chain=self.chain, @@ -590,10 +869,20 @@ def setup_firewall(self, enable_monitoring: bool = False) -> None: chain=LEGACY_CHAIN, policy=POLICY, family=LEGACY_FAMILY, table=LEGACY_TABLE ) + self.apply_user_rules() + + if default_drop: + self.ensure_default_drop(envelope) + else: + self.ensure_default_accept() + except Exception as e: logger.error('Failed to setup firewall: %s', e) raise NFTablesError(e) - logger.info('Firewall rules are configured') + logger.info( + 'Firewall rules are configured, default policy: %s', + POLICY_DROP if default_drop else POLICY, + ) def cleanup_legacy_rules(self, ssh: bool = False, dns: bool = False) -> None: """Cleans up all node-cli generated rules.""" @@ -631,17 +920,64 @@ def flush_chain(self, chain: str) -> None: raise NFTablesError('Flushing chain errored') +def firewall_default_drop_enabled() -> bool: + value = os.getenv(FIREWALL_DEFAULT_DROP_ENV, 'True') + return value.lower() not in ('false', '0', 'no', 'off') + + +def get_registered_base_port() -> Optional[int]: + """Base port for the envelope, taken from the node config. + + node_base_port is the port the node was registered with (saved by + skale-admin at registration and backfilled from the contracts on admin + restarts). schain_base_port is the fallback for passive and fair nodes, + where it holds the single hosted chain's base port - a valid anchor too. + """ + if not os.path.isfile(NODE_CONFIG_PATH): + return None + try: + node_config = read_json(NODE_CONFIG_PATH) + except (OSError, ValueError) as e: + logger.warning('Failed to read node config: %s', e) + return None + if not isinstance(node_config, dict): + logger.warning('Node config is malformed') + return None + for key in ('node_base_port', 'schain_base_port'): + value = node_config.get(key) + if isinstance(value, int) and not isinstance(value, bool) and value > 0: + return value + return None + + +def get_schain_ports_envelope() -> tuple[int, int]: + """Range of ports that can be allocated to sChains on this node.""" + env_value = os.getenv(SCHAIN_BASE_PORT_ENV) + if env_value: + try: + base_port = int(env_value) + except ValueError: + raise NFTablesError(f'{SCHAIN_BASE_PORT_ENV} must be an integer, got {env_value}') + else: + base_port = get_registered_base_port() or DEFAULT_NODE_BASE_PORT + if not MIN_SCHAIN_BASE_PORT <= base_port <= MAX_PORT - SCHAIN_PORTS_PER_NODE + 1: + raise NFTablesError(f'Invalid sChain base port {base_port}') + return base_port, base_port + SCHAIN_PORTS_PER_NODE - 1 + + def prepare_directories() -> None: logger.info('Prepare directories for nftables') os.makedirs(NFTABLES_CHAIN_FOLDER_PATH, exist_ok=True) create_user_config_path() -def configure_nftables(enable_monitoring: bool = False) -> None: +def configure_nftables(enable_monitoring: bool = False, keep_accept_policy: bool = False) -> None: prepare_directories() enable_nftables_service() nft_mgr = NFTablesManager() - nft_mgr.setup_firewall(enable_monitoring=enable_monitoring) + nft_mgr.setup_firewall( + enable_monitoring=enable_monitoring, keep_accept_policy=keep_accept_policy + ) ruleset = nft_mgr.get_base_ruleset() save_nftables_rules(ruleset) remove_legacy_saved_rules() diff --git a/node_cli/core/node.py b/node_cli/core/node.py index b1f22d86..55952421 100644 --- a/node_cli/core/node.py +++ b/node_cli/core/node.py @@ -34,6 +34,7 @@ CONTAINER_CONFIG_PATH, FILESTORAGE_MAPPING, LOG_PATH, + NODE_CONFIG_PATH, RESTORE_SLEEP_TIMEOUT, SCHAINS_MNT_DIR_REGULAR, SCHAINS_MNT_DIR_SINGLE_CHAIN, @@ -52,6 +53,7 @@ passive_skale, passive_fair, ) +from node_cli.core.nftables import get_registered_base_port from node_cli.migrations.focal_to_jammy import migrate as migrate_2_6 from node_cli.operations import ( cleanup_skale_op, @@ -79,6 +81,8 @@ error_exit, get_request, post_request, + read_json, + save_json, ) from node_cli.utils.meta import CliMetaManager from node_cli.utils.node_type import NodeType, NodeMode @@ -145,12 +149,40 @@ def register_node(name, p2p_ip, public_ip, port, domain_name): msg = TEXTS['node']['registered'] logger.info(msg) print(msg) + try: + save_registered_base_port(port) + logger.info('Reconfiguring firewall for the registered base port %d', port) + configure_nftables(enable_monitoring=get_settings().monitoring_containers) + except Exception: + # on-chain registration already succeeded - retrying register + # would fail, so the error must say the node is registered + logger.exception('Post-registration firewall reconfiguration failed') + error_exit( + 'Node is successfully registered in SKALE manager, but firewall ' + 'reconfiguration failed. Run < skale node configure-firewall > ' + 'to complete the setup', + exit_code=CLIExitCodes.OPERATION_EXECUTION_ERROR, + ) else: error_msg = payload logger.error(f'Registration error {error_msg}') error_exit(error_msg, exit_code=CLIExitCodes.BAD_API_RESPONSE) +def save_registered_base_port(port: int) -> None: + """Persist the node base port to the node config. + + Kept separate from schain_base_port, which holds an already-allocated + sChain port in passive mode. skale-admin saves node_base_port during + registration as well - this covers setups where the admin container + predates that behavior. + """ + node_config = read_json(NODE_CONFIG_PATH) if os.path.isfile(NODE_CONFIG_PATH) else {} + if node_config.get('node_base_port') != port: + node_config['node_base_port'] = port + save_json(NODE_CONFIG_PATH, node_config) + + @check_not_inited def init(config_file: str, node_type: NodeType) -> None: node_mode = NodeMode.ACTIVE @@ -217,9 +249,35 @@ def init_passive( time.sleep(TM_INIT_TIMEOUT) if not is_base_containers_alive(node_type=NodeType.SKALE, node_mode=node_mode): error_exit('Containers are not running', exit_code=CLIExitCodes.OPERATION_EXECUTION_ERROR) + enable_firewall_default_drop_when_port_available(settings) logger.info('Passive node initialized successfully') +def enable_firewall_default_drop_when_port_available( + settings, timeout: int = 300, interval: int = 5 +) -> None: + """Flip the firewall to default drop once admin saves the base port. + + Passive init configures nftables before skale-admin computes the mirrored + chain's base port, so the drop policy is deferred until the port is known. + """ + start = time.monotonic() + while time.monotonic() - start < timeout: + if get_registered_base_port() is not None: + configure_nftables(enable_monitoring=settings.monitoring_containers) + return + time.sleep(interval) + logger.warning( + 'Node base port is not available after %d seconds - firewall default ' + 'drop is postponed until the next node update', + timeout, + ) + print( + 'Firewall default drop policy is postponed: the chain base port is not ' + 'known yet. It will be applied on the next < skale node update-passive >' + ) + + @check_inited @check_user def update_passive(config_file: str) -> None: diff --git a/node_cli/operations/base.py b/node_cli/operations/base.py index 8b777ef9..21a219dd 100644 --- a/node_cli/operations/base.py +++ b/node_cli/operations/base.py @@ -223,7 +223,7 @@ def init_passive( if not settings.skip_docker_config: configure_docker() - configure_nftables(enable_monitoring=settings.monitoring_containers) + configure_nftables(enable_monitoring=settings.monitoring_containers, keep_accept_policy=True) prepare_host(env_type=settings.env_type) save_internal_settings(node_type=NodeType.SKALE, node_mode=NodeMode.PASSIVE) diff --git a/tests/cli/node_test.py b/tests/cli/node_test.py index a0c064f0..8e3ac3d8 100644 --- a/tests/cli/node_test.py +++ b/tests/cli/node_test.py @@ -58,7 +58,12 @@ def test_register_node(inited_node, resource_alloc, mocked_g_config): resp_mock = response_mock(requests.codes.ok, {'status': 'ok', 'payload': None}) - with mock.patch('node_cli.utils.decorators.is_node_inited', return_value=True): + with ( + mock.patch('node_cli.utils.decorators.is_node_inited', return_value=True), + mock.patch('node_cli.core.node.save_registered_base_port'), + mock.patch('node_cli.core.node.configure_nftables'), + mock.patch('node_cli.core.node.get_settings'), + ): result = run_command_mock( 'node_cli.utils.helper.requests.post', resp_mock, @@ -72,6 +77,37 @@ def test_register_node(inited_node, resource_alloc, mocked_g_config): ) # noqa +def test_register_node_firewall_failure(inited_node, resource_alloc, mocked_g_config): + """Post-registration firewall errors fail the command but report the registration.""" + resp_mock = response_mock(requests.codes.ok, {'status': 'ok', 'payload': None}) + with ( + mock.patch('node_cli.utils.decorators.is_node_inited', return_value=True), + mock.patch( + 'node_cli.core.node.save_registered_base_port', + side_effect=OSError('disk error'), + ), + mock.patch('node_cli.core.node.configure_nftables'), + mock.patch('node_cli.core.node.get_settings'), + ): + result = run_command_mock( + 'node_cli.utils.helper.requests.post', + resp_mock, + register_node, + ['--name', 'test-node', '--ip', '0.0.0.0', '--port', '8080', '-d', 'skale.test'], + ) + assert result.exit_code == CLIExitCodes.OPERATION_EXECUTION_ERROR.value + assert result.output == ( + 'Node registered in SKALE manager.\nFor more info run < skale node info >\n' + 'Command failed with following errors:\n' + '--------------------------------------------------\n' + 'Node is successfully registered in SKALE manager, but firewall ' + 'reconfiguration failed. Run < skale node configure-firewall > ' + 'to complete the setup\n' + '--------------------------------------------------\n' + f'You can find more info in {G_CONF_HOME}.skale/.skale-cli-log/debug-node-cli.log\n' + ) + + def test_register_node_with_error(inited_node, resource_alloc, mocked_g_config): resp_mock = response_mock( requests.codes.ok, @@ -93,7 +129,12 @@ def test_register_node_with_error(inited_node, resource_alloc, mocked_g_config): def test_register_node_with_prompted_ip(inited_node, resource_alloc, mocked_g_config): resp_mock = response_mock(requests.codes.ok, {'status': 'ok', 'payload': None}) - with mock.patch('node_cli.utils.decorators.is_node_inited', return_value=True): + with ( + mock.patch('node_cli.utils.decorators.is_node_inited', return_value=True), + mock.patch('node_cli.core.node.save_registered_base_port'), + mock.patch('node_cli.core.node.configure_nftables'), + mock.patch('node_cli.core.node.get_settings'), + ): result = run_command_mock( 'node_cli.utils.helper.requests.post', resp_mock, @@ -110,7 +151,12 @@ def test_register_node_with_prompted_ip(inited_node, resource_alloc, mocked_g_co def test_register_node_with_default_port(inited_node, resource_alloc, mocked_g_config): resp_mock = response_mock(requests.codes.ok, {'status': 'ok', 'payload': None}) - with mock.patch('node_cli.utils.decorators.is_node_inited', return_value=True): + with ( + mock.patch('node_cli.utils.decorators.is_node_inited', return_value=True), + mock.patch('node_cli.core.node.save_registered_base_port'), + mock.patch('node_cli.core.node.configure_nftables'), + mock.patch('node_cli.core.node.get_settings'), + ): result = run_command_mock( 'node_cli.utils.helper.requests.post', resp_mock, @@ -384,7 +430,9 @@ def test_maintenance_off(mocked_g_config): ) -def test_turn_off_maintenance_on(mocked_g_config, regular_user_conf, active_node_option, skale_active_settings): +def test_turn_off_maintenance_on( + mocked_g_config, regular_user_conf, active_node_option, skale_active_settings +): resp_mock = response_mock(requests.codes.ok, {'status': 'ok', 'payload': None}) with ( mock.patch('subprocess.run', new=subprocess_run_mock), @@ -415,7 +463,9 @@ def test_turn_off_maintenance_on(mocked_g_config, regular_user_conf, active_node assert result.exit_code == CLIExitCodes.UNSAFE_UPDATE -def test_turn_on_maintenance_off(mocked_g_config, regular_user_conf, active_node_option, skale_active_settings): +def test_turn_on_maintenance_off( + mocked_g_config, regular_user_conf, active_node_option, skale_active_settings +): resp_mock = response_mock(requests.codes.ok, {'status': 'ok', 'payload': None}) with ( mock.patch('subprocess.run', new=subprocess_run_mock), diff --git a/tests/core/nftables_test.py b/tests/core/nftables_test.py index b6490b08..4c3d6776 100644 --- a/tests/core/nftables_test.py +++ b/tests/core/nftables_test.py @@ -4,7 +4,13 @@ import nftables -from node_cli.core.nftables import NFTablesManager, Rule +import node_cli.core.nftables as nftables_core +from node_cli.core.nftables import ( + NFTablesError, + NFTablesManager, + Rule, + get_schain_ports_envelope, +) @pytest.fixture(scope='module') @@ -102,6 +108,50 @@ def test_create_chain_if_not_exists(mock_exists, mock_execute, nft_manager): mock_execute.assert_called_once() +@patch('nftables.Nftables.cmd') +def test_update_chain_policy_uses_given_policy(mock_cmd, nft_manager): + """Test that policy update applies the requested policy.""" + mock_cmd.return_value = (0, '', '') + with patch.object(NFTablesManager, 'chain_exists', return_value=True): + nft_manager.update_chain_policy(chain='skale', policy='drop') + assert mock_cmd.call_args[0][0] == 'add chain inet filter skale { policy drop ; }' + + mock_cmd.return_value = (1, '', 'some error') + with patch.object(NFTablesManager, 'chain_exists', return_value=True): + with pytest.raises(NFTablesError): + nft_manager.update_chain_policy(chain='skale', policy='drop') + + +@patch('nftables.Nftables.cmd') +def test_get_chain_policy(mock_cmd, nft_manager): + listing = { + 'nftables': [ + { + 'chain': { + 'family': 'inet', + 'table': 'filter', + 'name': 'skale', + 'hook': 'input', + 'policy': 'drop', + } + } + ] + } + mock_cmd.return_value = (0, json.dumps(listing), '') + assert nft_manager.get_chain_policy('skale') == 'drop' + + mock_cmd.return_value = (1, '', 'No such file or directory') + assert nft_manager.get_chain_policy('skale') is None + + mock_cmd.return_value = (0, 'not-json', '') + assert nft_manager.get_chain_policy('skale') is None + + # malformed but valid JSON must not raise - rollback depends on it + for output in ('[]', 'null', '{"nftables": [{"chain": null}]}', '{"nftables": "x"}'): + mock_cmd.return_value = (0, output, '') + assert nft_manager.get_chain_policy('skale') is None + + @pytest.mark.parametrize( 'rule_data', [ @@ -122,17 +172,501 @@ def test_add_rule(mock_exists, mock_execute, nft_manager, rule_data): @patch.object(NFTablesManager, 'execute_cmd') -def test_setup_firewall(mock_execute, nft_manager): +@patch.object(NFTablesManager, 'rule_exists') +def test_add_rule_udp_uses_udp_payload(mock_exists, mock_execute, nft_manager): + """Test that udp rules match udp dport, not tcp.""" + mock_exists.return_value = False + + nft_manager.add_rule(Rule(chain='INPUT', protocol='udp', first_port=53)) + expr = mock_execute.call_args[0][0]['nftables'][0]['add']['rule']['expr'] + assert expr[0]['match']['left']['payload'] == {'protocol': 'udp', 'field': 'dport'} + + +@patch.object(NFTablesManager, 'execute_cmd') +@patch.object(NFTablesManager, 'rule_exists') +def test_add_rule_icmpv6(mock_exists, mock_execute, nft_manager): + """Test icmpv6 rule addition.""" + mock_exists.return_value = False + + nft_manager.add_rule(Rule(chain='INPUT', protocol='icmpv6', icmp_type='nd-neighbor-solicit')) + expr = mock_execute.call_args[0][0]['nftables'][0]['add']['rule']['expr'] + assert expr[0]['match']['left']['payload'] == {'protocol': 'icmpv6', 'field': 'type'} + assert expr[0]['match']['right'] == 'nd-neighbor-solicit' + + +@patch('nftables.Nftables.cmd') +def test_get_dynamic_chain_port_ranges(mock_cmd, nft_manager): + """Test collection of port ranges covered by skale-admin chains.""" + listing = { + 'nftables': [ + {'chain': {'family': 'inet', 'table': 'filter', 'name': 'skale'}}, + {'chain': {'family': 'inet', 'table': 'filter', 'name': 'skale-test'}}, + { + 'rule': { + 'family': 'inet', + 'table': 'filter', + 'chain': 'skale', + 'expr': [ + { + 'match': { + 'op': '==', + 'left': {'payload': {'protocol': 'tcp', 'field': 'dport'}}, + 'right': 22, + } + }, + {'accept': None}, + ], + } + }, + { + 'rule': { + 'family': 'inet', + 'table': 'filter', + 'chain': 'skale-test', + 'expr': [ + { + 'match': { + 'op': '==', + 'left': {'payload': {'protocol': 'ip', 'field': 'saddr'}}, + 'right': '1.2.3.4', + } + }, + { + 'match': { + 'op': '==', + 'left': {'payload': {'protocol': 'tcp', 'field': 'dport'}}, + 'right': 10001, + } + }, + {'accept': None}, + ], + } + }, + { + 'rule': { + 'family': 'inet', + 'table': 'filter', + 'chain': 'skale-test', + 'expr': [ + { + 'match': { + 'op': '==', + 'left': {'payload': {'protocol': 'tcp', 'field': 'dport'}}, + 'right': {'range': [10000, 10063]}, + } + }, + {'drop': None}, + ], + } + }, + ] + } + mock_cmd.return_value = (0, json.dumps(listing), '') + assert nft_manager.get_dynamic_chain_port_ranges() == [('skale-test', 10000, 10063)] + + mock_cmd.return_value = (1, '', 'No such file or directory') + assert nft_manager.get_dynamic_chain_port_ranges() == [] + + +def test_validate_dynamic_ranges(nft_manager): + """Test envelope validation against dynamic chain ranges.""" + with patch.object( + NFTablesManager, + 'get_dynamic_chain_port_ranges', + return_value=[('skale-test', 10064, 10127)], + ): + nft_manager.validate_dynamic_ranges((10000, 18191)) + with pytest.raises(NFTablesError): + nft_manager.validate_dynamic_ranges((10128, 18191)) + + +def test_verify_critical_accepts(nft_manager): + with patch.object(NFTablesManager, 'rule_exists', return_value=True): + nft_manager.verify_critical_accepts() + with patch.object(NFTablesManager, 'rule_exists', return_value=False): + with pytest.raises(NFTablesError): + nft_manager.verify_critical_accepts() + + +def test_ensure_default_drop(nft_manager): + with patch.multiple( + NFTablesManager, + validate_dynamic_ranges=Mock(), + verify_critical_accepts=Mock(), + get_chain_policy=Mock(return_value='accept'), + update_chain_policy=Mock(), + ): + nft_manager.ensure_default_drop((10000, 18191)) + NFTablesManager.validate_dynamic_ranges.assert_called_once_with((10000, 18191)) + NFTablesManager.verify_critical_accepts.assert_called_once() + NFTablesManager.update_chain_policy.assert_called_once_with(chain='skale', policy='drop') + + with patch.multiple( + NFTablesManager, + validate_dynamic_ranges=Mock(), + verify_critical_accepts=Mock(), + get_chain_policy=Mock(return_value='drop'), + update_chain_policy=Mock(), + ): + nft_manager.ensure_default_drop((10000, 18191)) + NFTablesManager.update_chain_policy.assert_not_called() + + +def test_ensure_default_accept(nft_manager): + with patch.multiple( + NFTablesManager, + get_chain_policy=Mock(return_value='drop'), + update_chain_policy=Mock(), + ): + nft_manager.ensure_default_accept() + NFTablesManager.update_chain_policy.assert_called_once_with(chain='skale', policy='accept') + + with patch.multiple( + NFTablesManager, + get_chain_policy=Mock(return_value='accept'), + update_chain_policy=Mock(), + ): + nft_manager.ensure_default_accept() + NFTablesManager.update_chain_policy.assert_not_called() + + # unreadable policy must not skip the rollback + with patch.multiple( + NFTablesManager, + get_chain_policy=Mock(return_value=None), + update_chain_policy=Mock(), + ): + nft_manager.ensure_default_accept() + NFTablesManager.update_chain_policy.assert_called_once_with(chain='skale', policy='accept') + + +def test_remove_stale_envelope_rules(nft_manager): + stale_rule = { + 'handle': 7, + 'expr': [ + { + 'match': { + 'op': '==', + 'left': {'payload': {'protocol': 'tcp', 'field': 'dport'}}, + 'right': {'range': [10000, 18191]}, + } + }, + {'counter': None}, + {'accept': None}, + ], + } + current_rule = { + 'handle': 8, + 'expr': [ + { + 'match': { + 'op': '==', + 'left': {'payload': {'protocol': 'tcp', 'field': 'dport'}}, + 'right': {'range': [30000, 38191]}, + } + }, + {'counter': None}, + {'accept': None}, + ], + } + sgx_drop_rule = { + 'handle': 9, + 'expr': [ + { + 'match': { + 'op': '==', + 'left': {'payload': {'protocol': 'tcp', 'field': 'dport'}}, + 'right': {'range': [1026, 1031]}, + } + }, + {'counter': None}, + {'drop': None}, + ], + } + with patch.multiple( + NFTablesManager, + get_rules=Mock(return_value=[stale_rule, current_rule, sgx_drop_rule]), + delete_rule_by_handle=Mock(), + ): + nft_manager.remove_stale_envelope_rules((30000, 38191)) + NFTablesManager.delete_rule_by_handle.assert_called_once_with(7) + + +def test_get_schain_ports_envelope_default(monkeypatch, tmp_path): + monkeypatch.setattr(nftables_core, 'NODE_CONFIG_PATH', str(tmp_path / 'nonexistent.json')) + assert get_schain_ports_envelope() == (10000, 18191) + + +def test_get_schain_ports_envelope_env_override(monkeypatch): + monkeypatch.setenv('SCHAIN_BASE_PORT', '30000') + assert get_schain_ports_envelope() == (30000, 38191) + + monkeypatch.setenv('SCHAIN_BASE_PORT', 'not-a-port') + with pytest.raises(NFTablesError): + get_schain_ports_envelope() + + monkeypatch.setenv('SCHAIN_BASE_PORT', '65000') + with pytest.raises(NFTablesError): + get_schain_ports_envelope() + + monkeypatch.setenv('SCHAIN_BASE_PORT', '1000') + with pytest.raises(NFTablesError): + get_schain_ports_envelope() + + +def test_get_schain_ports_envelope_malformed_node_config(monkeypatch, tmp_path): + config_path = tmp_path / 'node_config.json' + monkeypatch.setattr(nftables_core, 'NODE_CONFIG_PATH', str(config_path)) + + for content in ( + 'not-json', + '[1, 2]', + '{"node_base_port": "not-a-port"}', + '{"node_base_port": true}', + '{"node_base_port": -1}', + ): + config_path.write_text(content) + # falls back to the default base port + assert get_schain_ports_envelope() == (10000, 18191) + + # an invalid primary value must not mask a valid fallback + config_path.write_text(json.dumps({'node_base_port': 'bad', 'schain_base_port': 20128})) + assert get_schain_ports_envelope() == (20128, 28319) + + +def test_get_schain_ports_envelope_from_node_config(monkeypatch, tmp_path): + config_path = tmp_path / 'node_config.json' + monkeypatch.setattr(nftables_core, 'NODE_CONFIG_PATH', str(config_path)) + + # active node: node_base_port saved at registration wins + config_path.write_text( + json.dumps({'node_id': 1, 'node_base_port': 20128, 'schain_base_port': 30000}) + ) + assert get_schain_ports_envelope() == (20128, 28319) + + # passive/fair node: only schain_base_port is present + config_path.write_text(json.dumps({'node_id': 1, 'schain_base_port': 20128})) + assert get_schain_ports_envelope() == (20128, 28319) + + +@patch.object(NFTablesManager, 'execute_cmd') +def test_setup_firewall(mock_execute, nft_manager, monkeypatch, tmp_path): """Test complete firewall setup.""" + monkeypatch.setattr(nftables_core, 'NODE_CONFIG_PATH', str(tmp_path / 'nonexistent.json')) + monkeypatch.setattr(nftables_core, 'NFTABLES_USER_CONFIG_PATH', str(tmp_path / 'user.conf')) with patch.multiple( NFTablesManager, table_exists=Mock(return_value=False), chain_exists=Mock(return_value=False), rule_exists=Mock(return_value=False), + verify_critical_accepts=Mock(), + get_dynamic_chain_port_ranges=Mock(return_value=[]), + get_chain_policy=Mock(return_value='accept'), + update_chain_policy=Mock(), ): nft_manager.setup_firewall() assert mock_execute.called + added_exprs = [ + call.args[0]['nftables'][0]['add']['rule']['expr'] + for call in mock_execute.call_args_list + if 'rule' in call.args[0]['nftables'][0].get('add', {}) + ] + envelope_exprs = [ + expr + for expr in added_exprs + if expr[0].get('match', {}).get('right') == {'range': [10000, 18191]} + and {'accept': None} in expr + ] + assert len(envelope_exprs) == 1 + icmpv6_exprs = [ + expr + for expr in added_exprs + if expr[0].get('match', {}).get('left', {}).get('payload', {}).get('protocol') + == 'icmpv6' + ] + assert len(icmpv6_exprs) == len(nftables_core.ICMPV6_ACCEPT_TYPES) + assert not any( + expr[0].get('match', {}).get('right') == 'source-quench' for expr in added_exprs + ) + + NFTablesManager.update_chain_policy.assert_any_call( + chain='INPUT', policy='accept', family='ip', table='filter' + ) + NFTablesManager.update_chain_policy.assert_any_call(chain='skale', policy='drop') + + +@patch.object(NFTablesManager, 'execute_cmd') +def test_setup_firewall_default_drop_disabled(mock_execute, nft_manager, monkeypatch, tmp_path): + """Test that FIREWALL_DEFAULT_DROP=False keeps the accept policy. + + Rollback must not be blocked by envelope validation. + """ + monkeypatch.setenv('FIREWALL_DEFAULT_DROP', 'False') + monkeypatch.setattr(nftables_core, 'NODE_CONFIG_PATH', str(tmp_path / 'nonexistent.json')) + monkeypatch.setattr(nftables_core, 'NFTABLES_USER_CONFIG_PATH', str(tmp_path / 'user.conf')) + with patch.multiple( + NFTablesManager, + table_exists=Mock(return_value=True), + chain_exists=Mock(return_value=True), + rule_exists=Mock(return_value=True), + validate_dynamic_ranges=Mock(), + ensure_default_drop=Mock(), + ensure_default_accept=Mock(), + update_chain_policy=Mock(), + ): + nft_manager.setup_firewall() + NFTablesManager.validate_dynamic_ranges.assert_not_called() + NFTablesManager.ensure_default_drop.assert_not_called() + NFTablesManager.ensure_default_accept.assert_called_once() + + +@patch.object(NFTablesManager, 'execute_cmd') +def test_setup_firewall_keep_accept_policy(mock_execute, nft_manager, monkeypatch, tmp_path): + """Test that keep_accept_policy skips the drop flip (passive init).""" + monkeypatch.setattr(nftables_core, 'NODE_CONFIG_PATH', str(tmp_path / 'nonexistent.json')) + monkeypatch.setattr(nftables_core, 'NFTABLES_USER_CONFIG_PATH', str(tmp_path / 'user.conf')) + with patch.multiple( + NFTablesManager, + table_exists=Mock(return_value=True), + chain_exists=Mock(return_value=True), + rule_exists=Mock(return_value=True), + validate_dynamic_ranges=Mock(), + ensure_default_drop=Mock(), + ensure_default_accept=Mock(), + update_chain_policy=Mock(), + ): + nft_manager.setup_firewall(keep_accept_policy=True) + NFTablesManager.ensure_default_drop.assert_not_called() + NFTablesManager.ensure_default_accept.assert_called_once() + + +def test_remove_source_quench_rule(nft_manager): + source_quench_rule = { + 'handle': 11, + 'expr': [ + { + 'match': { + 'left': {'payload': {'protocol': 'icmp', 'field': 'type'}}, + 'op': '==', + 'right': 'source-quench', + } + }, + {'counter': {'packets': 0, 'bytes': 0}}, + {'accept': None}, + ], + } + other_rule = { + 'handle': 12, + 'expr': [ + { + 'match': { + 'left': {'payload': {'protocol': 'icmp', 'field': 'type'}}, + 'op': '==', + 'right': 'destination-unreachable', + } + }, + {'counter': {'packets': 0, 'bytes': 0}}, + {'accept': None}, + ], + } + with patch.multiple( + NFTablesManager, + get_rules=Mock(return_value=[source_quench_rule, other_rule]), + delete_rule_by_handle=Mock(), + ): + nft_manager.remove_source_quench_rule() + NFTablesManager.delete_rule_by_handle.assert_called_once_with(11) + + with patch.multiple( + NFTablesManager, + get_rules=Mock(return_value=[other_rule]), + delete_rule_by_handle=Mock(), + ): + nft_manager.remove_source_quench_rule() + NFTablesManager.delete_rule_by_handle.assert_not_called() + + +def test_remove_misordered_udp_drop(nft_manager): + udp_drop = { + 'handle': 5, + 'expr': [ + { + 'match': { + 'left': {'payload': {'protocol': 'ip', 'field': 'protocol'}}, + 'op': '==', + 'right': 'udp', + } + }, + {'counter': {'packets': 0, 'bytes': 0}}, + {'drop': None}, + ], + } + udp_dns_accept = { + 'handle': 6, + 'expr': [ + { + 'match': { + 'op': '==', + 'left': {'payload': {'protocol': 'udp', 'field': 'dport'}}, + 'right': 53, + } + }, + {'counter': {'packets': 0, 'bytes': 0}}, + {'accept': None}, + ], + } + # drop shadows the accept -> removed + with patch.multiple( + NFTablesManager, + get_rules=Mock(return_value=[udp_drop, udp_dns_accept]), + delete_rule_by_handle=Mock(), + ): + nft_manager.remove_misordered_udp_drop() + NFTablesManager.delete_rule_by_handle.assert_called_once_with(5) + + # accept missing -> drop removed so the accept can land above it + with patch.multiple( + NFTablesManager, + get_rules=Mock(return_value=[udp_drop]), + delete_rule_by_handle=Mock(), + ): + nft_manager.remove_misordered_udp_drop() + NFTablesManager.delete_rule_by_handle.assert_called_once_with(5) + + # correct order -> untouched + with patch.multiple( + NFTablesManager, + get_rules=Mock(return_value=[udp_dns_accept, udp_drop]), + delete_rule_by_handle=Mock(), + ): + nft_manager.remove_misordered_udp_drop() + NFTablesManager.delete_rule_by_handle.assert_not_called() + + +@patch('nftables.Nftables.cmd') +def test_apply_user_rules(mock_cmd, nft_manager, monkeypatch, tmp_path): + user_conf = tmp_path / 'user.conf' + user_conf.write_text( + '# custom services\ntcp dport 5000 counter accept\n\ntcp dport 6000 counter accept\n' + ) + monkeypatch.setattr(nftables_core, 'NFTABLES_USER_CONFIG_PATH', str(user_conf)) + + mock_cmd.return_value = (0, '', '') + with patch.object( + NFTablesManager, + 'get_base_ruleset', + return_value='chain skale {\n\t\ttcp dport 5000 counter accept\n}', + ): + nft_manager.apply_user_rules() + + applied = [call.args[0] for call in mock_cmd.call_args_list] + assert applied == ['insert rule inet filter skale tcp dport 6000 counter accept'] + + mock_cmd.return_value = (1, '', 'syntax error') + with patch.object(NFTablesManager, 'get_base_ruleset', return_value=''): + with pytest.raises(NFTablesError): + nft_manager.apply_user_rules() + def test_invalid_protocol(nft_manager): """Test adding rule with invalid protocol."""