戰(zhàn):4 個(gè)運(yùn)維場景 + 1 套 Python 自動(dòng)化腳本模板)
AdGuardHome API 實(shí)戰(zhàn)4 個(gè)運(yùn)維場景 1 套 Python 自動(dòng)化腳本模板【免費(fèi)下載鏈接】AdGuardHomeNetwork-wide ads trackers blocking DNS server項(xiàng)目地址: https://gitcode.com/gh_mirrors/ad/AdGuardHome每天進(jìn) Web 界面手點(diǎn)一次“刷新規(guī)則”故障排查時(shí)翻不到想要的查詢記錄攔截率到底多少心里沒底——這些重復(fù)勞動(dòng)適合交給腳本。AdGuardHome API 走的是 HTTP 協(xié)議、Basic Auth 認(rèn)證用 Python 寫自動(dòng)化腳本巡檢、取證、監(jiān)控幾行代碼就能固化下來。從架構(gòu)圖能看出關(guān)鍵點(diǎn)Web 界面本身只是這套 REST API 的客戶端之一Query log、Stats、Filters 等模塊全部掛在同一個(gè) API 層上。也就是說你的腳本和界面走的是完全相同的通道界面上能做的API 都能做。 先用一個(gè) Session 打通 GET /control/status依賴只有一個(gè)。認(rèn)證信息在整個(gè)腳本里只寫一次掛到requests.Session上后續(xù)所有請求自動(dòng)攜帶不再重復(fù)傳。# 依賴安裝pip install requests import requests # 主機(jī)與認(rèn)證只寫一次后面所有小節(jié)復(fù)用這個(gè) session AGH_URL http://192.168.0.200:80 session requests.Session() session.auth (admin, admin) # 換成你的管理員賬號 session.headers.update({Content-Type: application/json}) # 首次調(diào)用拿服務(wù)器狀態(tài) status session.get(f{AGH_URL}/control/status, timeout10).json() print(status[protection_enabled], status[running], status[version])GET /control/status返回protection_enabled保護(hù)是否開啟、running、dns_addresses、dns_port、version等字段是任何巡檢邏輯的第一站。如果這一行跑通了剩下的場景只是換路徑和參數(shù)。過濾規(guī)則自動(dòng)開關(guān)與刷新目的規(guī)則源更新后不用再進(jìn)界面點(diǎn)刷新過濾開關(guān)也能被腳本控制。POST /control/filtering/config負(fù)責(zé)啟停interval是自動(dòng)刷新間隔單位小時(shí)POST /control/filtering/add_url追加規(guī)則源POST /control/filtering/refresh立即重新下載合并全部規(guī)則。# 開啟 DNS 過濾并設(shè)定 24 小時(shí)的自動(dòng)刷新間隔 session.post(f{AGH_URL}/control/filtering/config, json{enabled: True, interval: 24}) # 追加規(guī)則源url 換成你自己的指向本地鏡像更穩(wěn) session.post(f{AGH_URL}/control/filtering/add_url, json{url: http://192.168.0.255/lists/ads.txt, name: local-ads, whitelist: False}) # 立即重新下載并合并全部規(guī)則 session.post(f{AGH_URL}/control/filtering/refresh)這些規(guī)則源維護(hù)的就是上圖里 Rewrite、Lists 環(huán)節(jié)的數(shù)據(jù)請求在到達(dá)上游 DNS 之前就會(huì)被攔下。坑在refresh它是同步調(diào)用規(guī)則列表多時(shí)耗時(shí)遠(yuǎn)超 10 秒要單獨(dú)放寬超時(shí)add_url傳重復(fù)的url會(huì)被拒絕追加前先查現(xiàn)有列表。查詢?nèi)罩纠∨c清空目的排查“某域名為什么沒解析出來”時(shí)秒級把證據(jù)撈出來而不是在界面里翻頁。GET /control/querylog支持limit控制條數(shù)、search按域名或客戶端 IP 過濾。# 取最近 50 條并按域名過濾 log session.get(f{AGH_URL}/control/querylog, params{limit: 50, search: ads.example.com}).json() for item in log[data]: print(item[time], item[client], item[name], item[status], item[reason])返回體是{oldest, data}data每條含name域名、client來源 IP、upstream、status、reason、cached等字段status能區(qū)分是規(guī)則攔截、安全瀏覽替換還是搜索改寫。清空只有一個(gè)接口且不可恢復(fù)session.post(f{AGH_URL}/control/querylog_clear) # 建議先落盤備份再執(zhí)行查詢量與攔截量周期監(jiān)控目的讓攔截量從“看不見”變成周期采樣的數(shù)字。GET /control/stats一次拿到全部計(jì)數(shù)。stats session.get(f{AGH_URL}/control/stats).json() total stats[num_dns_queries] blocked stats[num_blocked_filtering] print(f查詢 {total}攔截 {blocked}攔截率 {blocked / max(total, 1):.1%})返回里還有num_replaced_safebrowsing、num_replaced_safesearch、num_replaced_rewrites等分項(xiàng)。監(jiān)控別寫死循環(huán)常駐進(jìn)程讓定時(shí)任務(wù)每 5 分鐘采樣一次、把結(jié)果追加到文件需要?dú)w零統(tǒng)計(jì)窗口時(shí)再調(diào)POST /control/stats_reset。上游 DNS 配置查詢與健康測試目的上游掛了時(shí)先確認(rèn)是誰的鍋。GET /control/dns_info讀出當(dāng)前配置POST /control/test_upstream_dns實(shí)測可達(dá)性。dns session.get(f{AGH_URL}/control/dns_info).json() # 測試一組上游是否可用bootstrap_dns 負(fù)責(zé)解析上游域名 result session.post(f{AGH_URL}/control/test_upstream_dns, json{bootstrap_dns: [1.1.1.1], upstream_dns: dns[upstream_dns]}).json() bad {k: v for k, v in result.items() if v ! OK} print(異常上游, bad or 無)返回體里每個(gè)上游對應(yīng)一個(gè)狀態(tài)OK表示可達(dá)其余是具體錯(cuò)誤文本比如i/o timeout。寫入側(cè)同理POST /control/dns_config提交改后的配置模式與上面的讀-測-寫一致不再重復(fù)。 把腳本做穩(wěn)重試、超時(shí)與狀態(tài)碼集中封裝前面為了省事沒寫異常處理。上生產(chǎn)前把重試、超時(shí)、401/404/5xx 集中封裝進(jìn)一個(gè)工具函數(shù)調(diào)用處保持干凈而不是散落滿屏 try/exceptimport time def call(session, method, path, retries3, **kwargs): 統(tǒng)一出口超時(shí)、重試、狀態(tài)碼判斷都在這里 kwargs.setdefault(timeout, 15) for attempt in range(retries): try: resp session.request(method, f{AGH_URL}{path}, **kwargs) except requests.exceptions.RequestException as e: # 連接失敗/超時(shí)退避重試最后一次直接拋出 if attempt retries - 1: time.sleep(2 * (attempt 1)) continue raise e if resp.status_code 401: # 憑證失效重試無意義 raise PermissionError(認(rèn)證失敗檢查 Basic Auth 賬號密碼) if resp.status_code 404: # 路徑拼錯(cuò)或 API 版本不匹配 raise ValueError(f接口不存在{path}) if resp.status_code 500 and attempt retries - 1: time.sleep(2 * (attempt 1)) # 5xx退避重試 continue resp.raise_for_status() return resp.json()調(diào)用處只剩一行call(session, GET, /control/status)。401 和 404 必須區(qū)分對待前者是憑證問題后者多半是路徑或版本不對都別盲目重試。接口全集以官方定義 openapi/openapi.yaml 為準(zhǔn)想擴(kuò)展別的操作直接查它。調(diào)度交給 cronWindows 用任務(wù)計(jì)劃程序腳本本身保持冪等、無狀態(tài)*/5 * * * * /usr/bin/python3 /opt/scripts/agh_monitor.py /var/log/agh_monitor.log 21完整腳本復(fù)制即可運(yùn)行把上面所有場景合成單main的可運(yùn)行版本憑證走環(huán)境變量不落代碼庫#!/usr/bin/env python3 AdGuardHome 日常運(yùn)維狀態(tài)巡檢、規(guī)則刷新、日志取證、統(tǒng)計(jì)采樣 import json import os import time import requests AGH_URL os.environ.get(AGH_URL, http://192.168.0.200:80) AUTH (os.environ.get(AGH_USER, admin), os.environ.get(AGH_PASS, )) def build_session(): 認(rèn)證只在這里出現(xiàn)一次 s requests.Session() s.auth AUTH s.headers.update({Content-Type: application/json}) return s def call(session, method, path, retries3, **kwargs): 統(tǒng)一出口超時(shí)、重試、狀態(tài)碼判斷 kwargs.setdefault(timeout, 15) for attempt in range(retries): try: resp session.request(method, f{AGH_URL}{path}, **kwargs) except requests.exceptions.RequestException as e: if attempt retries - 1: time.sleep(2 * (attempt 1)) continue raise e if resp.status_code 401: raise PermissionError(認(rèn)證失敗檢查 Basic Auth 憑證) if resp.status_code 404: raise ValueError(f接口不存在{path}) if resp.status_code 500 and attempt retries - 1: time.sleep(2 * (attempt 1)) continue resp.raise_for_status() return resp.json() def main(): session build_session() # 1) 狀態(tài)巡檢 print(json.dumps(call(session, GET, /control/status), ensure_asciiFalse)) # 2) 過濾規(guī)則確保開啟 觸發(fā)刷新規(guī)則多時(shí)放寬超時(shí) call(session, POST, /control/filtering/config, json{enabled: True, interval: 24}) call(session, POST, /control/filtering/refresh, timeout120) # 3) 日志取證最近 200 條落盤不直接清空 log call(session, GET, /control/querylog, params{limit: 200}) with open(/tmp/agh_querylog.json, w, encodingutf-8) as f: json.dump(log[data], f, ensure_asciiFalse, indent2) # 4) 統(tǒng)計(jì)采樣 print(json.dumps(call(session, GET, /control/stats), ensure_asciiFalse)) if __name__ __main__: main()? 上線前自查清單憑證經(jīng)環(huán)境變量或配置下發(fā)沒有硬編碼在腳本和倉庫里AGH_URL從腳本所在主機(jī)可達(dá)管理端口常只綁定內(nèi)網(wǎng)querylog_clear、stats_reset不可逆執(zhí)行前已有落盤備份cron 行帶日志重定向腳本冪等、可重復(fù)執(zhí)行refresh的超時(shí)按實(shí)際規(guī)則源數(shù)量調(diào)整120 秒起步兩個(gè)延伸方向把第 4 步的統(tǒng)計(jì)輸出接給 Prometheus 或 Grafana攔截率就有歷史曲線了照著 openapi/openapi.yaml 把 DNS 重寫、客戶端管理等接口納入同一個(gè)call()腳本骨架不用動(dòng)?!久赓M(fèi)下載鏈接】AdGuardHomeNetwork-wide ads trackers blocking DNS server項(xiàng)目地址: https://gitcode.com/gh_mirrors/ad/AdGuardHome創(chuàng)作聲明:本文部分內(nèi)容由AI輔助生成(AIGC),僅供參考