FM / python

/opt/dedrads/python

Name	Size	Mode	Actions
__pycache__/	-	0555	rm
audit_dns.py	5367	0555	edit dl rm
header.py	460	0555	edit dl rm
humanize.py	4988	0555	edit dl rm
sampledata.json	15854	0555	edit dl rm
__init__.py	0	0555	edit dl rm

Edit: /opt/dedrads/python/audit_dns.py (5367B)

#!/usr/lib/rads/venv/bin/python3
"""This will grab zone files and then check @ns1, @ns2, and @8.8.8.8 for
conflicting records. There are two main items that should be called from other
scripts: check_dns.server() and check_dns.user('userna5')"""
from platform import node
import subprocess
import sys
import yaml
from cpapis import whmapi1

def get_zones() -> list[dict[str, str]]:
    """Uses cPanel API to grab all zones. Used with server() but not user()"""
    try:
        zones = whmapi1('listzones', check=True)['data']['zone']
    except Exception as exc:
        print(exc, file=sys.stderr)
    return zones

def formatted_soa(soa: str | None):
    """Grabs Just the SOA part of what is returned when you dig SOA which is
    always in $3 column"""
    try:
        if soa != '' and soa is not None:
            return soa.split()[2]
    except Exception as exc:
        print(exc, file=sys.stderr)
        return soa
    return None

def clean_dig(dig_results: str):
    """strips leading and end spacing from dig"""
    if dig_results.strip() == '':
        return None
    return dig_results.strip()

def dig(*args: str) -> str:
    cmd = ['/usr/bin/dig']
    cmd.extend(args)
    return subprocess.check_output(cmd, text=True)

def dig_domain(localns, dom_dict: dict[str, str]):
    """Makes a dictionary of dig @ns1, @ns2, and @8.8.8.8 results for txt, A
    record, soa, and MX"""
    check_domain = dom_dict['domain'].strip()
    dnsinfo = [
        {
            'domain': check_domain,
            'localns': localns,
            localns[0]: {
                'A': clean_dig(dig(localns[0], check_domain, '+short')),
                'txt': clean_dig(
                    dig(localns[0], check_domain, 'txt', '+short')
                ),
                'soa': formatted_soa(
                    clean_dig(dig(localns[0], check_domain, 'soa', '+short'))
                ),
                'mx': clean_dig(dig(localns[0], check_domain, 'mx', '+short')),
            },
            localns[1]: {
                'A': clean_dig(dig(localns[1], check_domain, '+short')),
                'txt': clean_dig(
                    dig(localns[1], check_domain, 'txt', '+short')
                ),
                'soa': formatted_soa(
                    clean_dig(dig(localns[1], check_domain, 'soa', '+short'))
                ),
                'mx': clean_dig(dig(localns[1], check_domain, 'mx', '+short')),
            },
            'goog': {
                'A': clean_dig(dig('8.8.8.8', check_domain, '+short')),
                'txt': clean_dig(dig('8.8.8.8', check_domain, 'txt', '+short')),
                'soa': formatted_soa(
                    clean_dig(dig('8.8.8.8', check_domain, 'soa', '+short'))
                ),
                'mx': clean_dig(dig('8.8.8.8', check_domain, 'mx', '+short')),
            },
        }
    ]
    return dnsinfo

def server():
    """Call from the outside. Will give results for all domains on the server
    checks for mismatching in @ns and @ns2, missing at @ns and mismatching with
    public record."""
    if 'webhostinghub' in node():
        localns = ['@ns1.webhostinghub.com', '@ns2.webhostinghub.com']
    else:
        localns = ['@ns.inmotionhosting.com', '@ns2.inmotionhosting.com']
    zones = get_zones()
    zonesinfo = []
    for dom in zones:
        zonesinfo.append(dig_domain(localns, dom))
    return zonesinfo

def user(username):
    """Call from the outside. Will give results for all addon, parked, and main
    domain for one user Grab addons, main_domain, and parked domain from
    userdata files. Lots of this part stolen from account_review"""
    if 'webhostinghub' in node():
        localns = ['@ns1.webhostinghub.com', '@ns2.webhostinghub.com']
    else:
        localns = ['@ns.inmotionhosting.com', '@ns2.inmotionhosting.com']
    zones = user_zones(username)
    # print zones
    zonesinfo = []
    for dom in zones:
        zonesinfo.append(dig_domain(localns, dom))
    return zonesinfo

def user_zones(username):
    """Used by user() instead of cPanel API to pull zones quickly from userdata
    files Doing this via API would require list_accounts, listaddon, and
    listparked. 3 slow calls"""
    path = f'/var/cpanel/userdata/{username}/main'
    try:
        with open(path, encoding='utf-8') as f:
            data_map = yaml.load(f, Loader=yaml.SafeLoader)
    except (OSError, ValueError):
        sys.exit("userdata not found or corrupted")
    # create separate lists for each type of domain
    zones = []
    if data_map is not None:
        main_domain = data_map['main_domain']
        zones.append({'domain': main_domain, 'zonefile': f'{main_domain}.db'})
        for parked in data_map['parked_domains']:
            zones.append({'domain': parked, 'zonefile': f'{parked}.db'})
        for addon in data_map['addon_domains']:
            zones.append({'domain': addon, 'zonefile': f'{addon}.db'})
    return zones

def domain(domain_name):
    """Call from the outside. Used for a single domain."""
    if 'webhostinghub' in node():
        localns = ['@ns1.webhostinghub.com', '@ns2.webhostinghub.com']
    else:
        localns = ['@ns.inmotionhosting.com', '@ns2.inmotionhosting.com']
    dom = {'domain': domain_name, 'zone_file': f'{domain_name}.db'}
    # print zones
    zonesinfo = [dig_domain(localns, dom)]
    return zonesinfo

cmd: