225 lines
11 KiB
Python
225 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Les raisons du registre survivent dans les pare-feux et se corrigent sans recreer les regles."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import io
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from contextlib import redirect_stdout
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import jinja2
|
|
import yaml
|
|
|
|
RACINE = Path(__file__).resolve().parents[2]
|
|
sys.path.insert(0, str(RACINE / "scripts"))
|
|
|
|
import appliquer_opnsense as opn
|
|
import appliquer_proxmox_fw as pve
|
|
import resoudre_flux as flux
|
|
|
|
|
|
class ClusterFictif:
|
|
"""Une regle deja posee et une affectation ; toute ecriture de politique est refusee."""
|
|
hote = "cluster-fictif"
|
|
|
|
def __init__(self, ignorer=False):
|
|
self.ignorer = ignorer
|
|
self.ecritures = []
|
|
self.regle = {"pos": 3, "type": "in", "action": "ACCEPT", "source": "+t90-app",
|
|
"proto": "tcp", "dport": "5432", "enable": 0, "comment": "ancien"}
|
|
self.affectation = {"pos": 7, "type": "group", "action": "t90-pg", "enable": 1}
|
|
|
|
@staticmethod
|
|
def rate(rep):
|
|
return None
|
|
|
|
def __call__(self, chemin, methode="GET", corps=None):
|
|
objets = {"/cluster/firewall/groups/t90-pg/3": self.regle,
|
|
"/nodes/noeud/qemu/9001/firewall/rules/7": self.affectation}
|
|
if methode != "GET":
|
|
assert methode == "PUT" and chemin in objets and set(corps) == {"comment"}
|
|
self.ecritures.append((chemin, copy.deepcopy(corps)))
|
|
if not self.ignorer:
|
|
objets[chemin].update(corps)
|
|
return None
|
|
if chemin in objets:
|
|
return copy.deepcopy(objets[chemin])
|
|
lectures = {
|
|
"/cluster/firewall/ipset": [{"name": "t90-app"}],
|
|
"/cluster/firewall/ipset/t90-app": [{"cidr": "192.0.2.10"}],
|
|
"/cluster/firewall/groups": [{"group": "t90-pg"}],
|
|
"/cluster/firewall/groups/t90-pg": [self.regle],
|
|
"/cluster/resources?type=vm": [{"vmid": 9001, "node": "noeud"}],
|
|
"/nodes/noeud/qemu/9001/firewall/rules": [self.affectation],
|
|
"/nodes/noeud/qemu/9001/firewall/options": {"enable": 1, "policy_in": "REJECT"},
|
|
"/cluster/firewall/options": {"enable": 1, "policy_in": "ACCEPT"},
|
|
}
|
|
return copy.deepcopy(lectures[chemin])
|
|
|
|
|
|
class FrontiereFictive:
|
|
"""Les trois sortes de regles : des descriptions anciennes, des identites stables."""
|
|
def __init__(self, devis, ignorer=False):
|
|
self.ignorer = ignorer
|
|
self.ecritures = []
|
|
self.objets = {}
|
|
for objet, cle, champ, source in (
|
|
("filter", opn.cle_regle, "description", "regles"),
|
|
("source_nat", opn.cle_nat, "description", "nat"),
|
|
("d_nat", opn.cle_redirection, "descr", "redirections")):
|
|
self.objets[objet] = {"uuid": objet, champ: cle(devis[source][0]) + " — ancien",
|
|
"enabled": "0", "sequence": "42"}
|
|
|
|
def __call__(self, chemin, corps=None):
|
|
for objet, courant in self.objets.items():
|
|
prefixe = f"/api/firewall/{objet}/"
|
|
if not chemin.startswith(prefixe):
|
|
continue
|
|
commande = chemin[len(prefixe):].split("/")[0]
|
|
if commande == "search_rule":
|
|
# La regle humaine n'entre jamais dans le perimetre gere.
|
|
return {"rows": [copy.deepcopy(courant), {"uuid": "humaine", "description": "Personnel"}]}
|
|
if commande == "get_rule":
|
|
return {"rule": copy.deepcopy(courant)}
|
|
if commande == "set_rule":
|
|
champ = "descr" if objet == "d_nat" else "description"
|
|
assert set(corps) == {"rule"} and set(corps["rule"]) == {champ}
|
|
self.ecritures.append((chemin, copy.deepcopy(corps)))
|
|
if not self.ignorer:
|
|
courant.update(corps["rule"])
|
|
return {"result": "saved"}
|
|
if commande == "apply":
|
|
return {"status": "ok"}
|
|
raise AssertionError(f"Ecriture de politique inattendue : {chemin}")
|
|
if chemin == "/api/diagnostics/interface/getRoutes":
|
|
return [{"destination": "default"}]
|
|
if chemin in ("/api/firewall/alias/searchItem/", "/api/routes/routes/searchroute/"):
|
|
return {"rows": []}
|
|
if chemin == "/api/firewall/alias/reconfigure/":
|
|
return {"status": "ok"}
|
|
raise AssertionError(chemin)
|
|
|
|
|
|
class LibellesParefeu(unittest.TestCase):
|
|
def setUp(self):
|
|
self.sortie = redirect_stdout(io.StringIO())
|
|
self.sortie.__enter__()
|
|
self.addCleanup(self.sortie.__exit__, None, None, None)
|
|
|
|
def test_nft_genere_des_commentaires_natifs_et_protege_le_texte(self):
|
|
raison = 'Accès "base" via C:\\réseau\n' + 'é' * 150
|
|
declar = {"serveur_postgresql": [{"sens": "ingress", "port": 5432,
|
|
"protocole": "tcp", "pair": "serveur_keycloak",
|
|
"chiffrement": "tls-requis", "raison": raison}]}
|
|
data = {"all": {"children": {
|
|
"hotes_actifs": {"hosts": {"base": {"ansible_host": "192.0.2.20"},
|
|
"app": {"ansible_host": "192.0.2.10"}}},
|
|
"serveur_postgresql": {"hosts": {"base": {}}},
|
|
"serveur_keycloak": {"hosts": {"app": {}}}}}}
|
|
with tempfile.TemporaryDirectory() as tmp, \
|
|
patch.object(flux, "_inventaire_site", return_value=data), \
|
|
patch.object(flux, "_sources_admin_ssh_site", return_value=["192.0.2.0/24"]), \
|
|
patch.object(flux, "_ports_du_plan_site", return_value={}), \
|
|
patch("underlay.chemin", return_value=Path(tmp) / "underlay.yml"):
|
|
flux.generer_nftables(declar, site=True)
|
|
texte = (Path(tmp) / "flux-genere/base.nft").read_text()
|
|
regle = next(l for l in texte.splitlines() if "tcp dport 5432" in l)
|
|
commentaire = json.loads(regle.split(" comment ", 1)[1])
|
|
self.assertTrue(commentaire.startswith('serveur_postgresql: Accès "base" via C:\\réseau '))
|
|
self.assertLessEqual(len(commentaire.encode("utf-8")), 127)
|
|
self.assertTrue(commentaire.endswith("..."))
|
|
for ligne in texte.splitlines():
|
|
if ligne.strip().startswith(("ip ", "ip6 ", "iif", "oif", "ct ", "reject ")):
|
|
self.assertIn(" comment ", ligne)
|
|
|
|
def test_nft_repli_annote_toutes_ses_regles(self):
|
|
dossier = RACINE / "roles/nftables_baseline"
|
|
valeurs = yaml.safe_load((dossier / "defaults/main.yml").read_text())
|
|
valeurs.update(nftables_baseline_extra_tcp_ports=[8080], nftables_baseline_extra_udp_ports=[53])
|
|
env = jinja2.Environment()
|
|
env.filters["bool"] = bool
|
|
texte = env.from_string((dossier / "templates/nftables.conf.j2").read_text()).render(valeurs)
|
|
regles = [l for l in texte.splitlines() if l.strip().startswith(("iif ", "ct ", "ip ", "ip6 ", "tcp ", "udp "))]
|
|
self.assertGreaterEqual(len(regles), 8)
|
|
self.assertTrue(all(" comment " in l for l in regles))
|
|
|
|
def test_raison_vide_refusee(self):
|
|
fl = copy.deepcopy(flux.charger_flux())
|
|
fl["serveur_postgresql"][0]["raison"] = " \n "
|
|
with self.assertRaisesRegex(flux.ErreurFlux, "raison attendue"):
|
|
flux.valider(fl)
|
|
|
|
def devis_proxmox(self):
|
|
return {"blocs": [{"ipsets": {"t90-app": {"role": "application", "membres": ["192.0.2.10"]}},
|
|
"groupes": [{"nom": "t90-pg", "role": "serveur_postgresql", "regles": [
|
|
{"sens": "IN", "action": "ACCEPT", "source": "+t90-app", "proto": "tcp",
|
|
"dport": "5432", "via": "serveur_keycloak", "raison": "Persistance du SSO."}]}],
|
|
"affectations": [{"vmid": 9001, "hote": "base", "groupes": ["t90-pg"]}]}]}
|
|
|
|
def test_proxmox_corrige_en_place_puis_converge(self):
|
|
api = ClusterFictif()
|
|
devis = self.devis_proxmox()
|
|
with patch.object(pve, "_prefixes_retires", return_value=()):
|
|
p = pve.plan(api, devis)
|
|
self.assertFalse(p["groupes_majer"] or p["affect_majer"])
|
|
self.assertTrue(pve.afficher(p))
|
|
self.assertEqual(pve.appliquer(api, p), 0)
|
|
self.assertEqual(len(api.ecritures), 2)
|
|
self.assertEqual(api.regle["enable"], 0)
|
|
self.assertIn("Persistance du SSO.", api.regle["comment"])
|
|
self.assertFalse(pve.afficher(pve.plan(api, devis)))
|
|
|
|
def test_proxmox_detecte_une_ecriture_sans_effet(self):
|
|
api = ClusterFictif(ignorer=True)
|
|
with patch.object(pve, "_prefixes_retires", return_value=()):
|
|
self.assertEqual(pve.appliquer(api, pve.plan(api, self.devis_proxmox())), 1)
|
|
|
|
def devis_frontiere(self):
|
|
return {"alias": {}, "regles": [{"tenant": "exemple", "interface": "wan", "sens": "in",
|
|
"protocole": "tcp", "source": "any", "destination": "WEB", "ports": [443],
|
|
"role": "serveur_web_frontal", "raison": "Publication HTTPS des sites."}],
|
|
"nat": [{"tenant": "exemple", "interface": "wan", "source": "TENANT",
|
|
"destination": "any", "cible": "wanip"}],
|
|
"redirections": [{"interface": "wan", "protocole": "tcp", "port_public": 443,
|
|
"cible": "192.0.2.20", "port_local": 443, "role": "serveur_web_frontal"}]}
|
|
|
|
def test_opnsense_corrige_par_uuid_puis_converge(self):
|
|
devis = self.devis_frontiere()
|
|
api = FrontiereFictive(devis)
|
|
p = opn.plan(api, devis)
|
|
for categorie in ("regles", "nat", "rdr"):
|
|
self.assertFalse(p[categorie + "_creer"] or p[categorie + "_retirer"])
|
|
self.assertTrue(opn.afficher(p))
|
|
self.assertEqual(opn.appliquer(api, p), 0)
|
|
self.assertEqual(len(api.ecritures), 3)
|
|
self.assertTrue(all(x["enabled"] == "0" and x["sequence"] == "42" for x in api.objets.values()))
|
|
self.assertIn("Publication HTTPS des sites.", api.objets["filter"]["description"])
|
|
self.assertFalse(opn.afficher(opn.plan(api, devis)))
|
|
devis["regles"][0]["raison"] = "Nouvelle justification."
|
|
self.assertEqual(len(opn.plan(api, devis)["descriptions_majer"]), 1)
|
|
|
|
def test_opnsense_detecte_une_ecriture_sans_effet(self):
|
|
devis = self.devis_frontiere()
|
|
api = FrontiereFictive(devis, ignorer=True)
|
|
self.assertEqual(opn.appliquer(api, opn.plan(api, devis)), 1)
|
|
|
|
def test_opnsense_garde_la_cle_et_borne_les_octets(self):
|
|
r = self.devis_frontiere()["regles"][0]
|
|
r["raison"] = 'Accès "HTTPS"\n' + 'é' * 300
|
|
cle = opn.cle_regle(r)
|
|
texte = opn._corps_regle(r, cle)["description"]
|
|
self.assertEqual(texte.split(" — ")[0], cle)
|
|
self.assertIn('Accès "HTTPS"', texte)
|
|
self.assertNotIn("\n", texte)
|
|
self.assertLessEqual(len(texte.encode()), 255)
|
|
with self.assertRaises(flux.ErreurFlux):
|
|
opn._description("x" * 255, "role", "raison")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|