|
|
|
@ -72,7 +72,7 @@ class PortRange(str):
|
|
|
|
|
if start > end:
|
|
|
|
|
raise ValueError("invalid port range: start must be less than end")
|
|
|
|
|
|
|
|
|
|
return range(start, end)
|
|
|
|
|
return range(start, end + 1)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# Zones
|
|
|
|
@ -175,6 +175,7 @@ class SNat(RestrictiveBaseModel):
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class Nat(RestrictiveBaseModel):
|
|
|
|
|
protocols: set[str] = {"icmp", "udp", "tcp"}
|
|
|
|
|
src: AutoSet[IPv4Network | ZoneName]
|
|
|
|
|
dst: AutoSet[IPv4Network | ZoneName]
|
|
|
|
|
snat: SNat
|
|
|
|
@ -421,25 +422,25 @@ def parse_filter_rule(rule: Rule, zones: Zones) -> Iterator[nft.Rule]:
|
|
|
|
|
for attr, field in (("src", "saddr"), ("dst", "daddr")):
|
|
|
|
|
if getattr(rule, attr, None) is not None:
|
|
|
|
|
addrs, negated = zones_into_ip(getattr(rule, attr), zones)
|
|
|
|
|
addr_v4, addr_v6 = split_v4_v6(addrs)
|
|
|
|
|
addrs_v4, addrs_v6 = split_v4_v6(addrs)
|
|
|
|
|
|
|
|
|
|
if addr_v4:
|
|
|
|
|
if addrs_v4:
|
|
|
|
|
builder.add_v4(
|
|
|
|
|
nft.Match(
|
|
|
|
|
op=("!=" if negated else "=="),
|
|
|
|
|
left=nft.Payload(protocol="ip", field=field),
|
|
|
|
|
right=addr_v4,
|
|
|
|
|
right=addrs_v4,
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
else:
|
|
|
|
|
builder.disable_v4()
|
|
|
|
|
|
|
|
|
|
if addr_v6:
|
|
|
|
|
if addrs_v6:
|
|
|
|
|
builder.add_v6(
|
|
|
|
|
nft.Match(
|
|
|
|
|
op=("!=" if negated else "=="),
|
|
|
|
|
left=nft.Payload(protocol="ip6", field=field),
|
|
|
|
|
right=addr_v6,
|
|
|
|
|
right=addrs_v6,
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
else:
|
|
|
|
@ -547,16 +548,68 @@ def parse_filter(filter: Filter, zones: Zones) -> nft.Table:
|
|
|
|
|
return table
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def parse_nat(nat: list[Nat], zones: Zones) -> nft.Table:
|
|
|
|
|
chain = nft.Chain(
|
|
|
|
|
name="postrouting",
|
|
|
|
|
type="nat",
|
|
|
|
|
hook="postrouting",
|
|
|
|
|
policy="accept",
|
|
|
|
|
priority=100,
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
for entry in nat:
|
|
|
|
|
rule = nft.Rule()
|
|
|
|
|
|
|
|
|
|
for attr, field in (("src", "saddr"), ("dst", "daddr")):
|
|
|
|
|
addrs, negated = zones_into_ip(getattr(entry, attr), zones)
|
|
|
|
|
addrs_v4, _ = split_v4_v6(addrs)
|
|
|
|
|
|
|
|
|
|
if addrs_v4:
|
|
|
|
|
rule.stmts.append(
|
|
|
|
|
nft.Match(
|
|
|
|
|
op=("!=" if negated else "=="),
|
|
|
|
|
left=nft.Payload(protocol="ip", field=field),
|
|
|
|
|
right=addrs_v4,
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
rule.stmts.append(
|
|
|
|
|
nft.Match(
|
|
|
|
|
op="==",
|
|
|
|
|
left=nft.Payload(protocol="ip", field="protocol"),
|
|
|
|
|
right=entry.protocols,
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
rule.stmts.append(
|
|
|
|
|
nft.Snat(
|
|
|
|
|
addr=entry.snat.addr,
|
|
|
|
|
port=entry.snat.port,
|
|
|
|
|
persistent=entry.snat.persistent,
|
|
|
|
|
)
|
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
chain.rules.append(rule)
|
|
|
|
|
|
|
|
|
|
# Resulting table
|
|
|
|
|
table = nft.Table(name="nat", family="ip")
|
|
|
|
|
|
|
|
|
|
table.chains.append(chain)
|
|
|
|
|
|
|
|
|
|
return table
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def parse_firewall(firewall: Firewall, zones: Zones) -> nft.Ruleset:
|
|
|
|
|
# Tables
|
|
|
|
|
blacklist = parse_blacklist(firewall.blacklist, zones)
|
|
|
|
|
rpf = parse_reverse_path_filter(firewall.reverse_path_filter)
|
|
|
|
|
filter = parse_filter(firewall.filter, zones)
|
|
|
|
|
nat = parse_nat(firewall.nat, zones)
|
|
|
|
|
|
|
|
|
|
# Resulting ruleset
|
|
|
|
|
ruleset = nft.Ruleset(flush=True)
|
|
|
|
|
|
|
|
|
|
ruleset.tables.extend([blacklist, rpf, filter])
|
|
|
|
|
ruleset.tables.extend([blacklist, rpf, filter, nat])
|
|
|
|
|
|
|
|
|
|
return ruleset
|
|
|
|
|
|
|
|
|
|