138 lines
4.0 KiB
Python
138 lines
4.0 KiB
Python
"""
|
|
export HOSTED_ZONE_ID=<aws hosted zone ID>
|
|
export ROUTE53_RECORD=something.mydomain.com
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
from typing import TYPE_CHECKING
|
|
|
|
import boto3
|
|
|
|
if TYPE_CHECKING:
|
|
from mypy_boto3_route53 import Route53Client
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s.%(msecs)03d %(levelname)s %(module)s - %(funcName)s: %(message)s",
|
|
datefmt="%Y-%m-%d %H:%M:%S",
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
logger.setLevel(logging.INFO)
|
|
|
|
HOSTED_ZONE_ID = os.getenv("HOSTED_ZONE_ID")
|
|
ROUTE53_RECORD = os.getenv("ROUTE53_RECORD")
|
|
SKIP_IPV4 = os.getenv("SKIP_IPV4", "false").lower() == "true"
|
|
SKIP_IPV6 = os.getenv("SKIP_IPV6", "false").lower() == "true"
|
|
|
|
|
|
def get_ipv4() -> str:
|
|
result = subprocess.run(["curl", "-4", "ifconfig.me"], capture_output=True)
|
|
return result.stdout.decode()
|
|
|
|
|
|
def get_ipv6() -> str:
|
|
result = subprocess.run(["curl", "-6", "ifconfig.me"], capture_output=True)
|
|
return result.stdout.decode()
|
|
|
|
|
|
def update_ipv4(hosted_zone_id: str, record: str, public_ipv4: str):
|
|
client: Route53Client = boto3.client("route53")
|
|
try:
|
|
logger.info("Calling upsert for ipv4.")
|
|
client.change_resource_record_sets(
|
|
HostedZoneId=hosted_zone_id,
|
|
ChangeBatch={
|
|
"Comment": "Update Public Addresses",
|
|
"Changes": [
|
|
{
|
|
"Action": "UPSERT",
|
|
"ResourceRecordSet": {
|
|
"Name": f"{record}",
|
|
"Type": "A",
|
|
"TTL": 300,
|
|
"ResourceRecords": [{"Value": public_ipv4}],
|
|
},
|
|
}
|
|
],
|
|
},
|
|
)
|
|
logger.info(f"Successfully updated ipv4 for {record}")
|
|
except Exception as e:
|
|
logger.error(f"Error updating ipv4 for {record}.")
|
|
raise e
|
|
|
|
|
|
def update_ipv6(hosted_zone_id: str, record: str, public_ipv6: str):
|
|
client = boto3.client("route53")
|
|
try:
|
|
logger.info("Calling upsert for ipv6.")
|
|
client.change_resource_record_sets(
|
|
HostedZoneId=hosted_zone_id,
|
|
ChangeBatch={
|
|
"Comment": "Update Public Addresses",
|
|
"Changes": [
|
|
{
|
|
"Action": "UPSERT",
|
|
"ResourceRecordSet": {
|
|
"Name": f"{record}",
|
|
"Type": "AAAA",
|
|
"TTL": 300,
|
|
"ResourceRecords": [{"Value": public_ipv6}],
|
|
},
|
|
}
|
|
],
|
|
},
|
|
)
|
|
logger.info(f"Successfully updated ipv6 for {record}")
|
|
except Exception as e:
|
|
logger.error(f"Error updating ipv6 for {record}.")
|
|
raise e
|
|
|
|
|
|
def main():
|
|
if not HOSTED_ZONE_ID:
|
|
logger.error("HOSTED_ZONE_ID env var not found!")
|
|
exit(1)
|
|
|
|
if not ROUTE53_RECORD:
|
|
logger.error("ROUTE53_RECORD env var not found!")
|
|
exit(1)
|
|
|
|
logger.info(f"Attempting to update {ROUTE53_RECORD} from {HOSTED_ZONE_ID}.")
|
|
|
|
if SKIP_IPV4:
|
|
logger.warning("Skipping IPv4.")
|
|
else:
|
|
logger.info("Getting IPv4 address from ifconfig.me")
|
|
public_ipv4 = get_ipv4()
|
|
if not public_ipv4:
|
|
logger.error("Public IPv4 not found.")
|
|
exit(1)
|
|
logger.info(f"Public IPv4 is {public_ipv4}")
|
|
update_ipv4(
|
|
hosted_zone_id=HOSTED_ZONE_ID,
|
|
record=ROUTE53_RECORD,
|
|
public_ipv4=public_ipv4,
|
|
)
|
|
|
|
if SKIP_IPV6:
|
|
logger.warning("Skipping IPv6")
|
|
else:
|
|
logger.info("Getting IPv6 address from ifconfig.me")
|
|
public_ipv6 = get_ipv6()
|
|
if not public_ipv6:
|
|
logger.error("Public IPv6 not found.")
|
|
exit(1)
|
|
logger.info(f"Public IPv6 is {public_ipv6}")
|
|
update_ipv6(
|
|
hosted_zone_id=HOSTED_ZONE_ID,
|
|
record=ROUTE53_RECORD,
|
|
public_ipv6=public_ipv6,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|