95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
"""
|
|
Route53 DNS record updater.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
from typing import TYPE_CHECKING
|
|
|
|
import boto3
|
|
from dotenv import load_dotenv
|
|
|
|
try:
|
|
from mypy_boto3_route53 import Route53Client
|
|
except ImportError:
|
|
Route53Client = object # type: ignore
|
|
|
|
load_dotenv()
|
|
|
|
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__)
|
|
|
|
|
|
def _get_route53_client() -> Route53Client:
|
|
access_key = os.getenv("AWS_ACCESS_KEY_ID")
|
|
secret_key = os.getenv("AWS_SECRET_ACCESS_KEY")
|
|
if access_key and secret_key:
|
|
return boto3.client(
|
|
"route53",
|
|
aws_access_key_id=access_key,
|
|
aws_secret_access_key=secret_key,
|
|
)
|
|
return boto3.client("route53")
|
|
|
|
|
|
def update_ipv4(hosted_zone_id: str, record: str, public_ipv4: str) -> None:
|
|
logger.debug("Creating Route53 client for hosted zone %s", hosted_zone_id)
|
|
client: Route53Client = _get_route53_client()
|
|
logger.debug("Building ChangeBatch for IPv4: record=%s, ip=%s, ttl=300", record, public_ipv4)
|
|
try:
|
|
logger.info("Calling Route53 upsert for IPv4 on %s", record)
|
|
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("Successfully updated IPv4 for %s -> %s", record, public_ipv4)
|
|
except Exception as e:
|
|
logger.error("Error updating IPv4 for %s: %s", record, e)
|
|
raise e
|
|
|
|
|
|
def update_ipv6(hosted_zone_id: str, record: str, public_ipv6: str) -> None:
|
|
logger.debug("Creating Route53 client for hosted zone %s", hosted_zone_id)
|
|
client = _get_route53_client()
|
|
logger.debug("Building ChangeBatch for IPv6: record=%s, ip=%s, ttl=300", record, public_ipv6)
|
|
try:
|
|
logger.info("Calling Route53 upsert for IPv6 on %s", record)
|
|
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("Successfully updated IPv6 for %s -> %s", record, public_ipv6)
|
|
except Exception as e:
|
|
logger.error("Error updating IPv6 for %s: %s", record, e)
|
|
raise e
|