Untitled Paste

#!/usr/bin/python3
import os
import sys
import argparse
import subprocess
import logging

# Try imports to handle environments where dependencies might be missing during syntax checks
try:
from dns import resolver
from ipalib import api, errors
from ipapython import dnsutil
except ImportError as e:
print(f"Error: Missing required libraries. {e}")
print("Ensure 'python3-ipalib' and 'python3-dns' are installed.")
sys.exit(1)

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')
logger = logging.getLogger(__name__)

def parse_args():
parser = argparse.ArgumentParser(description="Manage FreeIPA DNS TXT records for ACME challenges.")

parser.add_argument("--action", choices=['add', 'delete'], help="Action to perform.")
parser.add_argument("--domain", help="The domain being validated (e.g., www.example.com).")
parser.add_argument("--validation", help="The validation string/token.")

# Auth options
parser.add_argument("--keytab", help="Path to Kerberos keytab for authentication.")
parser.add_argument("--principal", help="Kerberos principal to use with keytab (default: host/<fqdn>).")

args = parser.parse_args()

# Fallback to Certbot Environment Variables if CLI args are missing
if not args.domain and 'CERTBOT_DOMAIN' in os.environ:
args.domain = os.environ['CERTBOT_DOMAIN']

if not args.validation and 'CERTBOT_VALIDATION' in os.environ:
args.validation = os.environ['CERTBOT_VALIDATION']

if not args.action:
# If CERTBOT_AUTH_OUTPUT is set, Certbot is in cleanup phase
if 'CERTBOT_AUTH_OUTPUT' in os.environ:
args.action = 'delete'
elif 'CERTBOT_DOMAIN' in os.environ:
args.action = 'add'

if not args.domain or not args.validation or not args.action:
parser.error("Domain, Validation, and Action are required (via CLI args or Certbot ENV vars).")

return args

def kinit(keytab, principal=None):
"""Authenticate using a keytab if provided."""
cmd = ['kinit', '-kt', keytab]
if principal:
cmd.append(principal)

logger.info(f"Authenticating with keytab: {keytab}")
try:
subprocess.check_call(cmd)
except subprocess.CalledProcessError:
logger.error("Kerberos authentication failed.")
sys.exit(1)

def get_zone_and_name(domain):
"""Calculate the Zone and Relative Name for the ACME challenge."""
validation_domain = f'_acme-challenge.{domain}'

# Make absolute (trailing dot) to ensure correct parsing
fqdn = dnsutil.DNSName(validation_domain).make_absolute()

try:
# Find the authoritative zone for this name
zone_name = resolver.zone_for_name(fqdn)
zone = dnsutil.DNSName(zone_name)

# Calculate the relative name within that zone
name = fqdn.relativize(zone)

return zone, name
except Exception as e:
logger.error(f"Failed to determine DNS zone for {validation_domain}: {e}")
sys.exit(1)

def main():
args = parse_args()

# handle authentication
if args.keytab:
kinit(args.keytab, args.principal)

# Initialize FreeIPA API
api.bootstrap(context='cli')
api.finalize()

# Verify we have a session (env must have KRB5CCNAME or active ticket)
try:
api.Backend.rpcclient.connect()
except Exception as e:
logger.error(f"Failed to connect to FreeIPA API: {e}")
logger.error("Ensure you have a valid Kerberos ticket (run `kinit` or use --keytab).")
sys.exit(1)

zone, name = get_zone_and_name(args.domain)

logger.info(f"Zone: {zone}, Record: {name}, Value: {args.validation}")

try:
if args.action == 'add':
logger.info("Adding TXT record...")
try:
api.Command.dnsrecord_add(
zone,
name,
txtrecord=[args.validation],
dnsttl=60
)
logger.info("Successfully added record.")
except errors.DuplicateEntry:
logger.warning("Record already exists.")

elif args.action == 'delete':
logger.info("Deleting TXT record...")
try:
api.Command.dnsrecord_del(
zone,
name,
txtrecord=[args.validation]
)
logger.info("Successfully deleted record.")
except errors.NotFound:
logger.warning("Record not found, nothing to delete.")

except Exception as e:
logger.error(f"FreeIPA command failed: {e}")
sys.exit(1)

if __name__ == "__main__":
main()