agrynco
9e268b8ec3
Add scoped google api network policy, its refresh command, and the verified production operations runbook. Ref: IT-1074
63 lines
2.5 KiB
Python
63 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Render Google API IPv4 egress; never applies changes to Kubernetes."""
|
|
|
|
import argparse
|
|
import ipaddress
|
|
import json
|
|
from pathlib import Path
|
|
import urllib.request
|
|
|
|
|
|
def fetch(name):
|
|
with urllib.request.urlopen(f"https://www.gstatic.com/ipranges/{name}.json", timeout=20) as response:
|
|
return json.load(response)
|
|
|
|
|
|
def api_networks(google, cloud):
|
|
networks = [ipaddress.ip_network(p["ipv4Prefix"]) for p in google["prefixes"] if "ipv4Prefix" in p]
|
|
for prefix in cloud["prefixes"]:
|
|
if "ipv4Prefix" not in prefix:
|
|
continue
|
|
excluded = ipaddress.ip_network(prefix["ipv4Prefix"])
|
|
remaining = []
|
|
for network in networks:
|
|
if not network.overlaps(excluded):
|
|
remaining.append(network)
|
|
elif not network.subnet_of(excluded):
|
|
remaining.extend(network.address_exclude(excluded))
|
|
networks = remaining
|
|
return list(ipaddress.collapse_addresses(networks))
|
|
|
|
|
|
def render(google, cloud):
|
|
networks = api_networks(google, cloud)
|
|
if not networks or any(not network.is_global for network in networks):
|
|
raise ValueError("Expected nonempty public Google API IPv4 ranges")
|
|
labels = {"app.kubernetes.io/name": "hrynco-shared-notification",
|
|
"app.kubernetes.io/managed-by": "hrynco-shared-notification"}
|
|
return {
|
|
"apiVersion": "networking.k8s.io/v1", "kind": "NetworkPolicy",
|
|
"metadata": {"name": "web-worker-to-google-apis", "namespace": "shared-notification",
|
|
"labels": labels, "annotations": {
|
|
"hrynco.com/google-ranges-date": google["creationTime"],
|
|
"hrynco.com/cloud-ranges-date": cloud["creationTime"]}},
|
|
"spec": {"podSelector": {"matchLabels": labels, "matchExpressions": [
|
|
{"key": "app.kubernetes.io/component", "operator": "In", "values": ["web", "worker"]}]},
|
|
"policyTypes": ["Egress"], "egress": [{
|
|
"to": [{"ipBlock": {"cidr": str(network)}} for network in networks],
|
|
"ports": [{"protocol": "TCP", "port": 443}]}]},
|
|
}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
policy = render(fetch("goog"), fetch("cloud"))
|
|
args.output.write_text(json.dumps(policy, indent=2) + "\n", encoding="utf-8")
|
|
print(f"Rendered {len(policy['spec']['egress'][0]['to'])} IPv4 ranges; review before applying.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|