fix: few small tweaks to validation (#1219)

bootstrap/scripts/validation.py

@@ -1,4 +1,4 @@
-from email_validator import validate_email, EmailNotValidError
+from email_validator import validate_email, EmailNotValidError, EmailUndeliverableError
 from functools import wraps
 from shutil import which
 from typing import Callable, Optional
@@ -16,6 +16,7 @@ GLOBAL_CLI_TOOLS = ["age", "cloudflared", "flux", "sops", "jq", "kubeconform", "
 TALOS_CLI_TOOLS = ["talosctl", "talhelper"]
 K0S_CLI_TOOLS = ["k0sctl"]
 
+
 def required(*keys: str):
     def wrapper_outter(func: Callable):
         @wraps(func)
@@ -24,9 +25,12 @@ def required(*keys: str):
                 if data.get(key) is None:
                     raise ValueError(f"Missing required key {key}")
             return func(*[data[key] for key in keys], **kwargs)
+
         return wrapper
+
     return wrapper_outter
 
+
 def _validate_ip(ip: str) -> str:
     try:
         netaddr.IPAddress(ip)
@@ -34,6 +38,7 @@ def _validate_ip(ip: str) -> str:
         raise ValueError(f"Invalid IP address {ip}") from e
     return ip
 
+
 def _validate_cidr(cidr: str, family: int) -> str:
     try:
         network = netaddr.IPNetwork(cidr)
@@ -43,11 +48,13 @@ def _validate_cidr(cidr: str, family: int) -> str:
         raise ValueError(f"Invalid CIDR {cidr}") from e
     return cidr
 
+
 def _validate_distribution(distribution: str) -> None:
     if distribution not in DISTRIBUTIONS:
         raise ValueError(f"Invalid distribution {distribution}")
     return distribution
 
+
 def _validate_node(node: dict, node_cidr: str, distribution: str) -> None:
     if not node.get("name"):
         raise ValueError(f"A node is missing a name")
@@ -67,11 +74,13 @@ def _validate_node(node: dict, node_cidr: str, distribution: str) -> None:
         if result != 0:
             raise ValueError(f"Node {node.get('name')} port {port} is not open")
 
+
 def validate_python_version() -> None:
     required_version = (3, 11, 0)
     if sys.version_info < required_version:
         raise ValueError(f"Python version is below 3.11. Please upgrade.")
 
+
 @required("bootstrap_distribution")
 def validate_cli_tools(distribution: str, **_) -> None:
     distro = _validate_distribution(distribution)
@@ -85,10 +94,12 @@ def validate_cli_tools(distribution: str, **_) -> None:
         if not which(tool):
             raise ValueError(f"Missing required CLI tool {tool}")
 
+
 @required("bootstrap_distribution")
 def validate_distribution(distribution: str, **_) -> None:
     _validate_distribution(distribution)
 
+
 @required("bootstrap_github_username", "bootstrap_github_repository_name", "bootstrap_github_repository_branch")
 def validate_github(username: str, repository: str, branch: str, **_) -> None:
     try:
@@ -98,16 +109,19 @@ def validate_github(username: str, repository: str, branch: str, **_) -> None:
     except requests.exceptions.RequestException as e:
         raise ValueError(f"GitHub repository {username}/{repository} branch {branch} not found") from e
 
+
 @required("bootstrap_age_public_key")
 def validate_age(key: str, **_) -> None:
     if not re.match(r"^age1[a-z0-9]{0,58}$", key):
         raise ValueError(f"Invalid Age public key {key}")
 
+
 @required("bootstrap_timezone")
 def validate_timezone(timezone: str, **_) -> None:
-    if not timezone in available_timezones():
+    if timezone not in available_timezones():
         raise ValueError(f"Invalid timezone {timezone}")
 
+
 @required("bootstrap_ipv6_enabled", "bootstrap_cluster_cidr", "bootstrap_service_cidr")
 def validate_cluster_cidrs(ipv6_enabled: bool, cluster_cidr: str, service_cidr: str, **_) -> None:
     if not isinstance(ipv6_enabled, bool):
@@ -137,19 +151,25 @@ def validate_cluster_cidrs(ipv6_enabled: bool, cluster_cidr: str, service_cidr:
     _validate_cidr(cluster_cidr, 4)
     _validate_cidr(service_cidr, 4)
 
+
 @required("bootstrap_acme_email", "bootstrap_acme_production_enabled")
 def validate_acme_email(email: str, acme_production: bool, **_) -> None:
     try:
         validate_email(email)
+    except EmailUndeliverableError:
+        pass
     except EmailNotValidError as e:
         raise ValueError(f"Invalid ACME email {email}") from e
 
+
 @required("bootstrap_flux_github_webhook_token")
 def validate_flux_github_webhook_token(token: str, **_) -> None:
     if not re.match(r"^[a-zA-Z0-9]+$", token):
         raise ValueError(f"Invalid Flux GitHub webhook token ***")
 
-@required("bootstrap_cloudflare_domain", "bootstrap_cloudflare_token", "bootstrap_cloudflare_account_tag", "bootstrap_cloudflare_tunnel_secret", "bootstrap_cloudflare_tunnel_id")
+
+@required("bootstrap_cloudflare_domain", "bootstrap_cloudflare_token", "bootstrap_cloudflare_account_tag",
+          "bootstrap_cloudflare_tunnel_secret", "bootstrap_cloudflare_tunnel_id")
 def validate_cloudflare(domain: str, token: str, account_tag: str, tunnel_secret: str, tunnel_id: str, **_) -> None:
     try:
         cf = CloudFlare.CloudFlare(token=token)
@@ -159,7 +179,8 @@ def validate_cloudflare(domain: str, token: str, account_tag: str, tunnel_secret
     except CloudFlare.exceptions.CloudFlareAPIError as e:
         raise ValueError(f"Cloudflare domain {domain} not found or token does not have access to it") from e
     try:
-        request = requests.get(f"https://api.cloudflare.com/client/v4/accounts/{account_tag}/cfd_tunnel/{tunnel_id}", headers={"Authorization": f"Bearer {token}"})
+        request = requests.get(f"https://api.cloudflare.com/client/v4/accounts/{account_tag}/cfd_tunnel/{tunnel_id}",
+                               headers={"Authorization": f"Bearer {token}"})
         if request.status_code != 200:
             raise ValueError(f"Cloudflare tunnel for {account_tag} not found or token does not have access to it")
         if not json.loads(request.text)["success"]:
@@ -167,12 +188,16 @@ def validate_cloudflare(domain: str, token: str, account_tag: str, tunnel_secret
     except requests.exceptions.RequestException as e:
         raise ValueError(f"Cloudflare tunnel for {account_tag} not found or token does not have access to it") from e
 
+
 @required("bootstrap_dns_server")
 def validate_bootstrap_dns_server(dns_server: str, **_) -> None:
     _validate_ip(dns_server)
 
-@required("bootstrap_node_cidr", "bootstrap_kube_api_addr", "bootstrap_k8s_gateway_addr", "bootstrap_external_ingress_addr", "bootstrap_internal_ingress_addr")
+
-def validate_host_network(node_cidr: str, api_addr: str, gateway_addr: str, external_ingress_addr: str, internal_ingress_addr: str, **_) -> None:
+@required("bootstrap_node_cidr", "bootstrap_kube_api_addr", "bootstrap_k8s_gateway_addr",
+          "bootstrap_external_ingress_addr", "bootstrap_internal_ingress_addr")
+def validate_host_network(node_cidr: str, api_addr: str, gateway_addr: str, external_ingress_addr: str,
+                          internal_ingress_addr: str, **_) -> None:
     _validate_cidr(node_cidr, 4)
     _validate_ip(api_addr)
     _validate_ip(gateway_addr)
@@ -182,7 +207,8 @@ def validate_host_network(node_cidr: str, api_addr: str, gateway_addr: str, exte
     addrs = [api_addr, gateway_addr, external_ingress_addr, internal_ingress_addr]
     unique = set(addrs)
     if len(addrs) != len(unique):
-        raise ValueError(f"{api_addr} and {gateway_addr} and {external_ingress_addr} and {internal_ingress_addr} are not unique")
+        raise ValueError(
+            f"{api_addr} and {gateway_addr} and {external_ingress_addr} and {internal_ingress_addr} are not unique")
 
     node_cidr = netaddr.IPNetwork(node_cidr)
     if netaddr.IPAddress(api_addr) not in node_cidr:
@@ -190,9 +216,12 @@ def validate_host_network(node_cidr: str, api_addr: str, gateway_addr: str, exte
     if netaddr.IPAddress(gateway_addr) not in node_cidr:
         raise ValueError(f"Kubernetes gateway address {gateway_addr} is not in the node CIDR {node_cidr}")
     if netaddr.IPAddress(external_ingress_addr) not in node_cidr:
-        raise ValueError(f"Kubernetes external ingress address {external_ingress_addr} is not in the node CIDR {node_cidr}")
+        raise ValueError(
+            f"Kubernetes external ingress address {external_ingress_addr} is not in the node CIDR {node_cidr}")
     if netaddr.IPAddress(internal_ingress_addr) not in node_cidr:
-        raise ValueError(f"Kubernetes internal ingress address {internal_ingress_addr} is not in the node CIDR {node_cidr}")
+        raise ValueError(
+            f"Kubernetes internal ingress address {internal_ingress_addr} is not in the node CIDR {node_cidr}")
+
 
 @required("bootstrap_node_cidr", "bootstrap_nodes", "bootstrap_distribution")
 def validate_nodes(node_cidr: str, nodes: dict[list], distribution: str, **_) -> None:
@@ -210,10 +239,12 @@ def validate_nodes(node_cidr: str, nodes: dict[list], distribution: str, **_) ->
     for node in workers:
         _validate_node(node, node_cidr, distribution)
 
+
 def validate(data: dict) -> None:
     validate_python_version()
     validate_cli_tools(data)
     validate_distribution(data)
+    if not data.get("bootstrap_private_github_repo"):
         validate_github(data)
     validate_age(data)
     validate_timezone(data)