audit: +kb-audit-npm/creds/dns — расширение karpathy-style
- kb-audit-npm.py: NPM API → сверка с npm-proxy-hosts.md детектит новые/удалённые hosts + смену backend/SSL - kb-audit-creds.py: HEAD/GET-ping всех URL из credentials.md с fallback на GET при 501/405, skip embedded-creds URLs - kb-audit-dns.py: dig @8.8.8.8 и @10.0.0.1 для всех доменов NPM детектит NXDOMAIN + split-horizon Первый прогон нашёл: - NPM: 2 новых host (router/vpn.dttb.ru), 2 изменения (bitrix24 backend, git SSL) - Creds: все 12 URL reachable ✓ - DNS: itilegent.ru не резолвится (публичные записи протухли)
This commit is contained in:
dttb
158
scripts/kb-audit-creds.py
Executable file
158
scripts/kb-audit-creds.py
Executable file
@@ -0,0 +1,158 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
kb-audit-creds — берёт все URL из credentials.md, пингует каждый,
|
||||
фиксирует unreachable или нестандартные коды ответа.
|
||||
Пишет audit/YYYY-MM-DD-creds-drift.md.
|
||||
|
||||
Проверка только reachability (HTTP status). Не тестирует реальный логин.
|
||||
"""
|
||||
|
||||
import re
|
||||
import ssl
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
VAULT = Path(__file__).resolve().parent.parent
|
||||
CREDS = VAULT / "projects/dttb/credentials.md"
|
||||
OUT_DIR = VAULT / "audit"
|
||||
|
||||
_CTX = ssl.create_default_context()
|
||||
_CTX.check_hostname = False
|
||||
_CTX.verify_mode = ssl.CERT_NONE
|
||||
|
||||
|
||||
def extract_urls(text: str):
|
||||
"""URL в обычной записи + в markdown-таблице.
|
||||
Пропускаем URL с embedded credentials (http://user:pass@host) — они для git/curl,
|
||||
не для reachability-check.
|
||||
"""
|
||||
pattern = re.compile(r"https?://[a-zA-Z0-9._:@-]+(?:/[^\s`|<>\"')]*)?")
|
||||
seen = {}
|
||||
for m in pattern.finditer(text):
|
||||
url = m.group(0).rstrip("/.,;:)")
|
||||
# пропускаем URLs с embedded creds
|
||||
if re.match(r"https?://[^/]*@", url):
|
||||
continue
|
||||
if url not in seen:
|
||||
start = max(0, m.start() - 30)
|
||||
end = min(len(text), m.end() + 30)
|
||||
ctx = text[start:end].replace("\n", " ").replace("|", " ").strip()
|
||||
seen[url] = ctx[:80]
|
||||
return seen
|
||||
|
||||
|
||||
def _request(url: str, method: str, timeout: int):
|
||||
req = urllib.request.Request(url, method=method, headers={"User-Agent": "kb-audit/1.0"})
|
||||
r = urllib.request.urlopen(req, context=_CTX, timeout=timeout)
|
||||
return str(r.status), r.reason or ""
|
||||
|
||||
|
||||
def ping(url: str, timeout: int = 6) -> tuple[str, str]:
|
||||
"""Возвращает (status, detail). Пробуем HEAD, при 501/405 fallback на GET."""
|
||||
try:
|
||||
return _request(url, "HEAD", timeout)
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code in (501, 405):
|
||||
try:
|
||||
return _request(url, "GET", timeout)
|
||||
except urllib.error.HTTPError as e2:
|
||||
return str(e2.code), e2.reason or ""
|
||||
except Exception as e2:
|
||||
return "ERR", str(e2)[:60]
|
||||
return str(e.code), e.reason or ""
|
||||
except urllib.error.URLError as e:
|
||||
reason = str(e.reason)
|
||||
if "ssl" in reason.lower() or "certificate" in reason.lower():
|
||||
return "SSL", reason[:60]
|
||||
return "FAIL", reason[:60]
|
||||
except TimeoutError:
|
||||
return "TIMEOUT", ""
|
||||
except Exception as e:
|
||||
return "ERR", str(e)[:60]
|
||||
|
||||
|
||||
def classify(status: str) -> str:
|
||||
"""Status → категория для отчёта."""
|
||||
if status in ("200", "301", "302", "303", "307", "308"):
|
||||
return "✓ reachable"
|
||||
if status in ("401", "403"):
|
||||
return "✓ auth-required (сервер жив)"
|
||||
if status in ("404", "405"): # 405 на HEAD, 404 ок при ping host root
|
||||
return "⚠ 4xx (сервер жив, но путь/метод)"
|
||||
if status.startswith("5"):
|
||||
return "❌ 5xx server error"
|
||||
if status in ("FAIL", "ERR", "TIMEOUT", "SSL"):
|
||||
return "❌ недоступен"
|
||||
return f"? {status}"
|
||||
|
||||
|
||||
def main():
|
||||
today = date.today().isoformat()
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
out = OUT_DIR / f"{today}-creds-drift.md"
|
||||
|
||||
text = CREDS.read_text()
|
||||
urls = extract_urls(text)
|
||||
|
||||
if not urls:
|
||||
print("no URLs found in credentials.md", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
results = []
|
||||
for url, ctx in urls.items():
|
||||
status, detail = ping(url)
|
||||
category = classify(status)
|
||||
results.append((url, ctx, status, detail, category))
|
||||
|
||||
unreachable = [r for r in results if "❌" in r[4]]
|
||||
questionable = [r for r in results if "⚠" in r[4] or "?" in r[4]]
|
||||
ok = [r for r in results if "✓" in r[4]]
|
||||
|
||||
lines = [
|
||||
"---",
|
||||
f"date: {today}",
|
||||
"type: audit",
|
||||
"source: kb-audit-creds.py",
|
||||
"tags: [audit, creds, reachability]",
|
||||
"---",
|
||||
"",
|
||||
f"# Credentials reachability — {today}",
|
||||
"",
|
||||
f"Ping-проверка URL из [[../projects/dttb/credentials|credentials.md]].",
|
||||
f"Проверяется только reachability (HTTP status), не реальный логин.",
|
||||
"",
|
||||
f"- Всего URL: **{len(results)}**",
|
||||
f"- ✓ Reachable: {len(ok)} / ⚠ Questionable: {len(questionable)} / ❌ Unreachable: {len(unreachable)}",
|
||||
"",
|
||||
]
|
||||
|
||||
if unreachable:
|
||||
lines += ["## ❌ Недоступные (проверить: сервер упал? URL поменялся?)", "",
|
||||
"| URL | Status | Detail | Контекст |", "|---|---|---|---|"]
|
||||
for url, ctx, st, det, _ in unreachable:
|
||||
lines.append(f"| `{url}` | {st} | {det[:40]} | {ctx[:50]} |")
|
||||
lines.append("")
|
||||
|
||||
if questionable:
|
||||
lines += ["## ⚠ Нестандартный ответ", "",
|
||||
"| URL | Status | Detail |", "|---|---|---|"]
|
||||
for url, ctx, st, det, _ in questionable:
|
||||
lines.append(f"| `{url}` | {st} | {det[:40]} |")
|
||||
lines.append("")
|
||||
|
||||
lines += ["## ✓ Все ответили нормально", "",
|
||||
"| URL | Status | Категория |", "|---|---|---|"]
|
||||
for url, ctx, st, det, cat in ok:
|
||||
lines.append(f"| `{url}` | {st} | {cat} |")
|
||||
lines += ["", "---", "*Автоматически через `scripts/kb-audit-creds.py`.*"]
|
||||
|
||||
out.write_text("\n".join(lines))
|
||||
print(f"creds drift: {out}")
|
||||
print(f" ok: {len(ok)} / questionable: {len(questionable)} / unreachable: {len(unreachable)}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
136
scripts/kb-audit-dns.py
Executable file
136
scripts/kb-audit-dns.py
Executable file
@@ -0,0 +1,136 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
kb-audit-dns — резолвит все домены из NPM API через публичный DNS (8.8.8.8),
|
||||
сравнивает с локальным резолвом, детектит NXDOMAIN и расхождения.
|
||||
Пишет audit/YYYY-MM-DD-dns-drift.md.
|
||||
"""
|
||||
|
||||
import json
|
||||
import ssl
|
||||
import subprocess
|
||||
import sys
|
||||
import urllib.request
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
NPM_URL = "https://npm.dttb.ru"
|
||||
NPM_USER = "it5870@yandex.ru"
|
||||
NPM_PASS = "1qaz!QAZ"
|
||||
|
||||
VAULT = Path(__file__).resolve().parent.parent
|
||||
OUT_DIR = VAULT / "audit"
|
||||
|
||||
_CTX = ssl.create_default_context()
|
||||
_CTX.check_hostname = False
|
||||
_CTX.verify_mode = ssl.CERT_NONE
|
||||
|
||||
|
||||
def fetch_npm_domains():
|
||||
"""Берём домены из NPM API (вместо парсинга markdown — свежее)."""
|
||||
req = urllib.request.Request(
|
||||
f"{NPM_URL}/api/tokens",
|
||||
data=json.dumps({"identity": NPM_USER, "secret": NPM_PASS}).encode(),
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
token = json.loads(urllib.request.urlopen(req, context=_CTX, timeout=10).read())["token"]
|
||||
req2 = urllib.request.Request(
|
||||
f"{NPM_URL}/api/nginx/proxy-hosts",
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
)
|
||||
hosts = json.loads(urllib.request.urlopen(req2, context=_CTX, timeout=10).read())
|
||||
domains = set()
|
||||
for h in hosts:
|
||||
for d in h.get("domain_names", []):
|
||||
domains.add(d)
|
||||
return sorted(domains)
|
||||
|
||||
|
||||
def dig(domain: str, server: str = "8.8.8.8") -> list[str]:
|
||||
"""dig +short @server domain → список IP (пусто если NXDOMAIN/недоступен)."""
|
||||
try:
|
||||
r = subprocess.run(
|
||||
["dig", f"@{server}", "+short", "+time=3", "+tries=1", domain, "A"],
|
||||
capture_output=True, text=True, timeout=8,
|
||||
)
|
||||
return [l.strip() for l in r.stdout.splitlines() if l.strip() and not l.startswith(";")]
|
||||
except (subprocess.TimeoutExpired, FileNotFoundError):
|
||||
return []
|
||||
|
||||
|
||||
def main():
|
||||
today = date.today().isoformat()
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
out = OUT_DIR / f"{today}-dns-drift.md"
|
||||
|
||||
try:
|
||||
domains = fetch_npm_domains()
|
||||
except Exception as e:
|
||||
print(f"NPM fetch fail: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
if len(domains) < 3:
|
||||
print(f"safety abort: domains {len(domains)} < 3", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
results = []
|
||||
for d in domains:
|
||||
public = dig(d, "8.8.8.8")
|
||||
local = dig(d, "10.0.0.1") # OpenWrt router
|
||||
results.append((d, public, local))
|
||||
|
||||
no_public = [r for r in results if not r[1]]
|
||||
split = [r for r in results if r[1] and r[2] and set(r[1]) != set(r[2])]
|
||||
no_local = [r for r in results if not r[2]]
|
||||
|
||||
lines = [
|
||||
"---",
|
||||
f"date: {today}",
|
||||
"type: audit",
|
||||
"source: kb-audit-dns.py",
|
||||
"tags: [audit, dns]",
|
||||
"---",
|
||||
"",
|
||||
f"# DNS resolve audit — {today}",
|
||||
"",
|
||||
f"Резолвим все домены из NPM через публичный DNS (8.8.8.8) и локальный роутер (10.0.0.1).",
|
||||
"",
|
||||
f"- Всего доменов: **{len(domains)}**",
|
||||
f"- NXDOMAIN на 8.8.8.8: {len(no_public)} / пустой ответ локально: {len(no_local)} / split-horizon: {len(split)}",
|
||||
"",
|
||||
]
|
||||
|
||||
if no_public:
|
||||
lines += ["## ❌ NXDOMAIN / не резолвится на 8.8.8.8 (публичный DNS)", ""]
|
||||
lines += ["| Домен | Локальный IP |", "|---|---|"]
|
||||
for d, _, loc in no_public:
|
||||
lines.append(f"| `{d}` | {','.join(loc) or '(тоже нет)'} |")
|
||||
lines += [""]
|
||||
|
||||
if split:
|
||||
lines += ["## ⚠ Split-horizon — разные IP снаружи и внутри", "",
|
||||
"Это нормально для *.dttb.ru (внешний Let's Encrypt IP vs локальный 10.0.0.195). Но неожиданный split может быть багом.", ""]
|
||||
lines += ["| Домен | Публичный (8.8.8.8) | Локальный (10.0.0.1) |", "|---|---|---|"]
|
||||
for d, pub, loc in split:
|
||||
lines.append(f"| `{d}` | {','.join(pub)} | {','.join(loc)} |")
|
||||
lines += [""]
|
||||
|
||||
if no_local:
|
||||
lines += ["## ⚠ Пустой локальный резолв (роутер не знает)", ""]
|
||||
for d, pub, _ in no_local:
|
||||
lines.append(f"- `{d}` (публичный: {','.join(pub) or '-'})")
|
||||
lines += [""]
|
||||
|
||||
lines += ["## Полная таблица резолва", "",
|
||||
"| Домен | 8.8.8.8 | 10.0.0.1 |", "|---|---|---|"]
|
||||
for d, pub, loc in results:
|
||||
lines.append(f"| `{d}` | {','.join(pub) or '—'} | {','.join(loc) or '—'} |")
|
||||
lines += ["", "---", "*Автоматически через `scripts/kb-audit-dns.py`.*"]
|
||||
|
||||
out.write_text("\n".join(lines))
|
||||
print(f"dns drift: {out}")
|
||||
print(f" no_public: {len(no_public)} / split: {len(split)} / no_local: {len(no_local)}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
181
scripts/kb-audit-npm.py
Executable file
181
scripts/kb-audit-npm.py
Executable file
@@ -0,0 +1,181 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
kb-audit-npm — сверяет NPM API с npm-proxy-hosts.md.
|
||||
Детектит: новые proxy hosts, удалённые, смена backend IP/порт, SSL-дрейф, enabled-флаг.
|
||||
Пишет в audit/YYYY-MM-DD-npm-drift.md.
|
||||
"""
|
||||
|
||||
import json
|
||||
import re
|
||||
import ssl
|
||||
import sys
|
||||
import urllib.request
|
||||
from datetime import date
|
||||
from pathlib import Path
|
||||
|
||||
NPM_URL = "https://npm.dttb.ru"
|
||||
NPM_USER = "it5870@yandex.ru"
|
||||
NPM_PASS = "1qaz!QAZ"
|
||||
|
||||
VAULT = Path(__file__).resolve().parent.parent
|
||||
NPM_FILE = VAULT / "projects/dttb/npm-proxy-hosts.md"
|
||||
OUT_DIR = VAULT / "audit"
|
||||
|
||||
_CTX = ssl.create_default_context()
|
||||
_CTX.check_hostname = False
|
||||
_CTX.verify_mode = ssl.CERT_NONE
|
||||
|
||||
|
||||
def npm_api(path: str, token: str = None, method: str = "GET", body: dict = None):
|
||||
headers = {"Content-Type": "application/json"} if body else {}
|
||||
if token:
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
req = urllib.request.Request(
|
||||
f"{NPM_URL}{path}",
|
||||
data=json.dumps(body).encode() if body else None,
|
||||
headers=headers,
|
||||
method=method,
|
||||
)
|
||||
return json.loads(urllib.request.urlopen(req, context=_CTX, timeout=15).read())
|
||||
|
||||
|
||||
def fetch_live_hosts():
|
||||
token = npm_api("/api/tokens", method="POST", body={"identity": NPM_USER, "secret": NPM_PASS})["token"]
|
||||
hosts = npm_api("/api/nginx/proxy-hosts", token=token)
|
||||
result = {}
|
||||
for h in hosts:
|
||||
hid = str(h["id"])
|
||||
result[hid] = {
|
||||
"domains": ",".join(h.get("domain_names", [])),
|
||||
"backend": f"{h.get('forward_host','?')}:{h.get('forward_port','?')}",
|
||||
"scheme": h.get("forward_scheme", "http"),
|
||||
"ssl": bool(h.get("certificate_id")),
|
||||
"enabled": bool(h.get("enabled")),
|
||||
"websockets": bool(h.get("allow_websocket_upgrade")),
|
||||
}
|
||||
return result
|
||||
|
||||
|
||||
def parse_kb_hosts():
|
||||
"""Таблица proxy-hosts. Поля: # | Домен | Backend | SSL | Forced | WSS | HTTP/2 | Назначение."""
|
||||
text = NPM_FILE.read_text()
|
||||
result = {}
|
||||
for m in re.finditer(
|
||||
r"^\|\s*(\d+)\s*\|\s*`([^`]+)`\s*\|\s*([^\|]+?)\s*\|\s*([^\|]+?)\s*\|",
|
||||
text,
|
||||
re.MULTILINE,
|
||||
):
|
||||
hid, domain, backend, ssl_col = m.groups()
|
||||
backend = backend.strip()
|
||||
# убираем "(HTTPS)" суффикс
|
||||
backend_clean = re.sub(r"\s*\(HTTPS\)\s*$", "", backend)
|
||||
result[hid] = {
|
||||
"domain": domain.strip(),
|
||||
"backend": backend_clean.strip(),
|
||||
"ssl": "✅" in ssl_col,
|
||||
}
|
||||
return result
|
||||
|
||||
|
||||
def compare(live: dict, kb: dict):
|
||||
live_ids = set(live.keys())
|
||||
kb_ids = set(kb.keys())
|
||||
added = sorted(live_ids - kb_ids, key=int)
|
||||
removed = sorted(kb_ids - live_ids, key=int)
|
||||
changed = []
|
||||
for hid in sorted(live_ids & kb_ids, key=int):
|
||||
l, k = live[hid], kb[hid]
|
||||
diffs = []
|
||||
# backend сравниваем "host:port"
|
||||
if l["backend"].lower() != k["backend"].lower():
|
||||
diffs.append(f"backend: KB=`{k['backend']}` → live=`{l['backend']}`")
|
||||
# ssl
|
||||
if l["ssl"] != k["ssl"]:
|
||||
diffs.append(f"ssl: KB={'✓' if k['ssl'] else '✗'} → live={'✓' if l['ssl'] else '✗'}")
|
||||
if diffs:
|
||||
changed.append((hid, l["domains"], diffs))
|
||||
return added, removed, changed
|
||||
|
||||
|
||||
def main():
|
||||
today = date.today().isoformat()
|
||||
OUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
out = OUT_DIR / f"{today}-npm-drift.md"
|
||||
|
||||
try:
|
||||
live = fetch_live_hosts()
|
||||
except Exception as e:
|
||||
print(f"NPM API fail: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
if len(live) < 3:
|
||||
print(f"safety abort: live {len(live)} < 3", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
|
||||
kb = parse_kb_hosts()
|
||||
added, removed, changed = compare(live, kb)
|
||||
|
||||
lines = [
|
||||
"---",
|
||||
f"date: {today}",
|
||||
"type: audit",
|
||||
"source: kb-audit-npm.py",
|
||||
"tags: [audit, drift, npm]",
|
||||
"---",
|
||||
"",
|
||||
f"# NPM drift audit — {today}",
|
||||
"",
|
||||
f"Сверка [[../projects/dttb/npm-proxy-hosts|npm-proxy-hosts.md]] с NPM API ({NPM_URL}).",
|
||||
"",
|
||||
f"- Живых proxy hosts: **{len(live)}**",
|
||||
f"- В KB: **{len(kb)}**",
|
||||
f"- Совпадений: {len(set(live) & set(kb))} / новых: {len(added)} / удалённых из NPM: {len(removed)} / с изменениями: {len(changed)}",
|
||||
"",
|
||||
]
|
||||
|
||||
if added:
|
||||
lines += ["## ⚠ Новые hosts (в NPM есть, в KB нет)", "",
|
||||
"| ID | Домены | Backend | SSL | Enabled |",
|
||||
"|---|---|---|---|---|"]
|
||||
for hid in added:
|
||||
h = live[hid]
|
||||
lines.append(f"| {hid} | `{h['domains']}` | `{h['backend']}` | {'✓' if h['ssl'] else '-'} | {'on' if h['enabled'] else 'off'} |")
|
||||
lines.append("")
|
||||
|
||||
if removed:
|
||||
lines += ["## 🗑 Удалены из NPM (но есть в KB)", "",
|
||||
"| ID | Домен из KB | Backend KB |", "|---|---|---|"]
|
||||
for hid in removed:
|
||||
k = kb[hid]
|
||||
lines.append(f"| {hid} | `{k['domain']}` | `{k['backend']}` |")
|
||||
lines.append("")
|
||||
|
||||
if changed:
|
||||
lines += ["## 🔄 Изменения (ID совпадает, но что-то сменилось)", ""]
|
||||
for hid, domains, diffs in changed:
|
||||
lines.append(f"### #{hid} `{domains}`")
|
||||
for d in diffs:
|
||||
lines.append(f"- {d}")
|
||||
lines.append("")
|
||||
|
||||
if not (added or removed or changed):
|
||||
lines += ["## ✓ NPM полностью совпадает с npm-proxy-hosts.md", ""]
|
||||
|
||||
lines += [
|
||||
"## Полный живой список",
|
||||
"",
|
||||
"| ID | Домены | Backend | SSL | Enabled |",
|
||||
"|---|---|---|---|---|",
|
||||
]
|
||||
for hid in sorted(live.keys(), key=int):
|
||||
h = live[hid]
|
||||
lines.append(f"| {hid} | `{h['domains']}` | `{h['backend']}` | {'✓' if h['ssl'] else '-'} | {'on' if h['enabled'] else 'off'} |")
|
||||
lines += ["", "---", "*Автоматически через `scripts/kb-audit-npm.py`.*"]
|
||||
|
||||
out.write_text("\n".join(lines))
|
||||
print(f"npm drift: {out}")
|
||||
print(f" added: {len(added)} / removed: {len(removed)} / changed: {len(changed)}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user