summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorJohn Estabrook <jestabro@vyos.io>2025-07-20 14:56:50 -0500
committerJohn Estabrook <jestabro@vyos.io>2025-07-20 18:58:26 -0500
commit30ecc2de64682522ca099bfd23c1535aebcddcd6 (patch)
treeea3388fba278eac423f97c6791447c866ec68b1b
parentd9f12bfe584c6568239db15d23eaffbd9f5b5a5c (diff)
downloadvyos-1x-30ecc2de64682522ca099bfd23c1535aebcddcd6.tar.gz
vyos-1x-30ecc2de64682522ca099bfd23c1535aebcddcd6.zip
T7645: formatting
-rwxr-xr-xsrc/conf_mode/nat64.py135
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)