diff options
| author | John Estabrook <jestabro@vyos.io> | 2025-07-20 14:56:50 -0500 |
|---|---|---|
| committer | John Estabrook <jestabro@vyos.io> | 2025-07-20 18:58:26 -0500 |
| commit | 30ecc2de64682522ca099bfd23c1535aebcddcd6 (patch) | |
| tree | ea3388fba278eac423f97c6791447c866ec68b1b | |
| parent | d9f12bfe584c6568239db15d23eaffbd9f5b5a5c (diff) | |
| download | vyos-1x-30ecc2de64682522ca099bfd23c1535aebcddcd6.tar.gz vyos-1x-30ecc2de64682522ca099bfd23c1535aebcddcd6.zip | |
T7645: formatting
| -rwxr-xr-x | src/conf_mode/nat64.py | 135 |
1 files changed, 67 insertions, 68 deletions
diff --git a/src/conf_mode/nat64.py b/src/conf_mode/nat64.py index 2c8d17cde..0ea594e1a 100755 --- a/src/conf_mode/nat64.py +++ b/src/conf_mode/nat64.py @@ -38,16 +38,16 @@ from vyos.utils.process import run airbag.enable() -INSTANCE_REGEX = re.compile(r"instance-(\d+)") -JOOL_CONFIG_DIR = "/run/jool" +INSTANCE_REGEX = re.compile(r'instance-(\d+)') +JOOL_CONFIG_DIR = '/run/jool' def get_config(config: Config | None = None) -> ConfigDict: if config is None: config = Config() - base = ["nat64"] - nat64 = config.get_config_dict(base, key_mangling=("-", "_"), get_first_key=True) + base = ['nat64'] + nat64 = config.get_config_dict(base, key_mangling=('-', '_'), get_first_key=True) config_diff = get_config_diff(config) # get_config_dict returns an instance of ConfigDict @@ -61,11 +61,11 @@ def verify(nat64) -> None: config_diff = getattr(nat64, 'config_diff') - check_kmod(["jool"]) - base_src = ["nat64", "source", "rule"] + check_kmod(['jool']) + base_src = ['nat64', 'source', 'rule'] # Load in existing instances so we can destroy any unknown - lines = cmd("jool instance display --csv").splitlines() + lines = cmd('jool instance display --csv').splitlines() for _, instance, _ in csv.reader(lines): match = INSTANCE_REGEX.fullmatch(instance) if not match: @@ -73,69 +73,68 @@ def verify(nat64) -> None: continue num = match.group(1) - rules = nat64.setdefault("source", {}).setdefault("rule", {}) + rules = nat64.setdefault('source', {}).setdefault('rule', {}) # Mark it for deletion if num not in rules: - rules[num] = {"deleted": True} + rules[num] = {'deleted': True} continue # If the user changes the mode, recreate the instance else Jool fails with: # Jool error: Sorry; you can't change an instance's framework for now. - if config_diff.is_node_changed(base_src + [f"instance-{num}", "mode"]): - rules[num]["recreate"] = True + if config_diff.is_node_changed(base_src + [f'instance-{num}', 'mode']): + rules[num]['recreate'] = True # If the user changes the pool6, recreate the instance else Jool fails with: # Jool error: Sorry; you can't change a NAT64 instance's pool6 for now. - if dict_search("source.prefix", rules[num]) and config_diff.is_node_changed( - base_src + [num, "source", "prefix"], + if dict_search('source.prefix', rules[num]) and config_diff.is_node_changed( + base_src + [num, 'source', 'prefix'], ): - rules[num]["recreate"] = True + rules[num]['recreate'] = True if not nat64: # nothing left to do return - - if dict_search("source.rule", nat64): + if dict_search('source.rule', nat64): # Ensure only 1 netfilter instance per namespace nf_rules = filter( - lambda i: "deleted" not in i and i.get('mode') == "netfilter", - nat64["source"]["rule"].values(), + lambda i: 'deleted' not in i and i.get('mode') == 'netfilter', + nat64['source']['rule'].values(), ) next(nf_rules, None) # Discard the first element if next(nf_rules, None) is not None: raise ConfigError( - "Jool permits only 1 NAT64 netfilter instance (per network namespace)" + 'Jool permits only 1 NAT64 netfilter instance (per network namespace)' ) - for rule, instance in nat64["source"]["rule"].items(): - if "deleted" in instance: + for rule, instance in nat64['source']['rule'].items(): + if 'deleted' in instance: continue # Verify that source.prefix is set and is a /96 - if not dict_search("source.prefix", instance): - raise ConfigError(f"Source NAT64 rule {rule} missing source prefix") - src_prefix = IPv6Network(instance["source"]["prefix"]) + if not dict_search('source.prefix', instance): + raise ConfigError(f'Source NAT64 rule {rule} missing source prefix') + src_prefix = IPv6Network(instance['source']['prefix']) if src_prefix.prefixlen != 96: - raise ConfigError(f"Source NAT64 rule {rule} source prefix must be /96") + raise ConfigError(f'Source NAT64 rule {rule} source prefix must be /96') if (int(src_prefix[0]) & int(IPv6Address('0:0:0:0:ff00::'))) != 0: raise ConfigError( f'Source NAT64 rule {rule} source prefix is not RFC6052-compliant: ' 'bits 64 to 71 (9th octet) must be zeroed' ) - pools = dict_search("translation.pool", instance) + pools = dict_search('translation.pool', instance) if pools: for num, pool in pools.items(): - if "address" not in pool: + if 'address' not in pool: raise ConfigError( - f"Source NAT64 rule {rule} translation pool " - f"{num} missing address/prefix" + f'Source NAT64 rule {rule} translation pool ' + f'{num} missing address/prefix' ) - if "port" not in pool: + if 'port' not in pool: raise ConfigError( - f"Source NAT64 rule {rule} translation pool " - f"{num} missing port(-range)" + f'Source NAT64 rule {rule} translation pool ' + f'{num} missing port(-range)' ) @@ -147,54 +146,54 @@ def generate(nat64) -> None: os.makedirs(JOOL_CONFIG_DIR, exist_ok=True) # pylint: disable=too-many-nested-blocks - if dict_search("source.rule", nat64): - for rule, instance in nat64["source"]["rule"].items(): - if "deleted" in instance: + if dict_search('source.rule', nat64): + for rule, instance in nat64['source']['rule'].items(): + if 'deleted' in instance: # Delete the unused instance file - os.unlink(os.path.join(JOOL_CONFIG_DIR, f"instance-{rule}.json")) + os.unlink(os.path.join(JOOL_CONFIG_DIR, f'instance-{rule}.json')) continue - name = f"instance-{rule}" + name = f'instance-{rule}' config = { - "instance": name, - "framework": "netfilter", - "global": { - "pool6": instance["source"]["prefix"], - "manually-enabled": "disable" not in instance, + 'instance': name, + 'framework': 'netfilter', + 'global': { + 'pool6': instance['source']['prefix'], + 'manually-enabled': 'disable' not in instance, }, # "bib": [], } - if "description" in instance: - config["comment"] = instance["description"] + if 'description' in instance: + config['comment'] = instance['description'] - if dict_search("translation.pool", instance): + if dict_search('translation.pool', instance): pool4 = [] # mark mark = '' - if dict_search("match.mark", instance): - mark = instance["match"]["mark"] + if dict_search('match.mark', instance): + mark = instance['match']['mark'] - for pool in instance["translation"]["pool"].values(): - if "disable" in pool: + for pool in instance['translation']['pool'].values(): + if 'disable' in pool: continue - protos = pool.get("protocol", {}).keys() or ("tcp", "udp", "icmp") + protos = pool.get('protocol', {}).keys() or ('tcp', 'udp', 'icmp') for proto in protos: obj = { - "protocol": proto.upper(), - "prefix": pool["address"], - "port range": pool["port"], + 'protocol': proto.upper(), + 'prefix': pool['address'], + 'port range': pool['port'], } if mark: - obj["mark"] = int(mark) - if "description" in pool: - obj["comment"] = pool["description"] + obj['mark'] = int(mark) + if 'description' in pool: + obj['comment'] = pool['description'] pool4.append(obj) if pool4: - config["pool4"] = pool4 + config['pool4'] = pool4 write_file(f'{JOOL_CONFIG_DIR}/{name}.json', json_write(config, indent=2)) @@ -204,30 +203,30 @@ def apply(nat64) -> None: unload_kmod(['jool']) return - if dict_search("source.rule", nat64): + if dict_search('source.rule', nat64): # Deletions first to avoid conflicts - for rule, instance in nat64["source"]["rule"].items(): - if not any(k in instance for k in ("deleted", "recreate")): + for rule, instance in nat64['source']['rule'].items(): + if not any(k in instance for k in ('deleted', 'recreate')): continue - ret = run(f"jool instance remove instance-{rule}") + ret = run(f'jool instance remove instance-{rule}') if ret != 0: raise ConfigError( - f"Failed to remove nat64 source rule {rule} (jool instance instance-{rule})" + f'Failed to remove nat64 source rule {rule} (jool instance instance-{rule})' ) # Now creations - for rule, instance in nat64["source"]["rule"].items(): - if "deleted" in instance: + for rule, instance in nat64['source']['rule'].items(): + if 'deleted' in instance: continue - name = f"instance-{rule}" - ret = run(f"jool -i {name} file handle {JOOL_CONFIG_DIR}/{name}.json") + name = f'instance-{rule}' + ret = run(f'jool -i {name} file handle {JOOL_CONFIG_DIR}/{name}.json') if ret != 0: - raise ConfigError(f"Failed to set jool instance {name}") + raise ConfigError(f'Failed to set jool instance {name}') -if __name__ == "__main__": +if __name__ == '__main__': try: c = get_config() verify(c) |
