fix(ops): allow google api egress and document recovery
Add scoped google api network policy, its refresh command, and the verified production operations runbook. Ref: IT-1074
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user