أتمتة DNS Drift Detection: اكتشف أي تعديل على DNS قبل ما الموقع يقع

import dns.resolver
import dns.message
import dns.query
import json
import sys
import os
from pathlib import Path
from urllib.request import Request, urlopen

DOMAINS = ["example.com", "api.example.com", "mail.example.com"]
RECORD_TYPES = ["A", "AAAA", "MX", "TXT", "NS", "CNAME"]
SNAPSHOT_PATH = Path("dns_snapshot.json")
SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK_URL"]

def get_authoritative_ns(domain: str) -> list[str]:
    parent = ".".join(domain.split(".")[-2:])
    answer = dns.resolver.resolve(parent, "NS")
    return sorted([str(rr.target).rstrip(".") for rr in answer])

def query_authoritative(domain: str, rtype: str, ns_host: str) -> list[str]:
    try:
        ns_ip = dns.resolver.resolve(ns_host, "A")[0].to_text()
        q = dns.message.make_query(domain, rtype)
        r = dns.query.udp(q, ns_ip, timeout=5)
        return sorted([str(a).strip() for a in r.answer])
    except Exception:
        return []

def snapshot_current() -> dict:
    result = {}
    for domain in DOMAINS:
        result[domain] = {}
        try:
            ns_list = get_authoritative_ns(domain)
        except Exception:
            continue
        primary_ns = ns_list[0]
        for rtype in RECORD_TYPES:
            answers = query_authoritative(domain, rtype, primary_ns)
            if answers:
                result[domain][rtype] = answers
    return result

def diff(old: dict, new: dict) -> list[str]:
    changes = []
    all_domains = set(old) | set(new)
    for d in sorted(all_domains):
        old_d = old.get(d, {})
        new_d = new.get(d, {})
        for rtype in sorted(set(old_d) | set(new_d)):
            if old_d.get(rtype) != new_d.get(rtype):
                changes.append(f"{d} [{rtype}] : {old_d.get(rtype)} -> {new_d.get(rtype)}")
    return changes

def notify_slack(changes: list[str]):
    text = ":rotating_light: DNS Drift Detected:\n" + "\n".join(changes)
    body = json.dumps({"text": text}).encode()
    req = Request(SLACK_WEBHOOK, data=body, headers={"Content-Type": "application/json"})
    urlopen(req, timeout=10)

if __name__ == "__main__":
    new_snap = snapshot_current()
    if SNAPSHOT_PATH.exists():
        old_snap = json.loads(SNAPSHOT_PATH.read_text())
        changes = diff(old_snap, new_snap)
        if changes:
            notify_slack(changes)
            print("CHANGES DETECTED:\n" + "\n".join(changes))
            sys.exit(1)
    SNAPSHOT_PATH.write_text(json.dumps(new_snap, indent=2, sort_keys=True))
    print("OK — snapshot up to date.")

name: DNS Drift Check
on:
  schedule:
    - cron: "*/5 * * * *"
  workflow_dispatch:

jobs:
  check:
    runs-on: ubuntu-latest
    permissions:
      contents: write
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install dnspython==2.6.1
      - name: Run DNS check
        env:
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}
        run: python dns_check.py
      - name: Commit baseline if first run
        if: github.event_name == 'workflow_dispatch'
        run: |
          git config user.name "dns-bot"
          git config user.email "bot@users.noreply.github.com"
          git add dns_snapshot.json
          git diff --cached --quiet || git commit -m "Update DNS baseline"
          git push