nearxos
9fc35e86fe
- Updated .env.example to include Asterisk templates path. - Modified docker-compose.yml to mount the templates directory. - Enhanced backend Dockerfile to copy templates into the container. - Introduced Asterisk diagnostics functionality in asterisk_profiles.py, allowing for baseline checks and diagnostics reporting. - Integrated Asterisk diagnostics into the device diagnostics workflow in graph.py. - Added formatting for Asterisk baseline drift reports in diagnostic_format.py. - Updated SKILL.md to document new config baseline drift feature for Asterisk. This commit enhances the system's capabilities for managing Asterisk configurations and diagnostics, improving overall troubleshooting processes.
294 lines
8.4 KiB
Python
294 lines
8.4 KiB
Python
"""Compare live Asterisk config against golden templates on disk."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_manifest_cache: dict[str, dict] = {}
|
|
_manifest_mtime: float = -1.0
|
|
|
|
_PLACEHOLDER_RE = re.compile(r"\{\{([A-Z0-9_]+)\}\}")
|
|
|
|
|
|
@dataclass
|
|
class BaselineCheck:
|
|
kind: str # required_line | required_substring
|
|
expected: str
|
|
found: bool
|
|
ok: bool
|
|
|
|
def as_dict(self) -> dict:
|
|
return {
|
|
"kind": self.kind,
|
|
"expected": self.expected,
|
|
"found": self.found,
|
|
"ok": self.ok,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class BaselineFileResult:
|
|
path: str
|
|
description: str
|
|
ok: bool
|
|
missing: bool = False
|
|
checks: list[BaselineCheck] = field(default_factory=list)
|
|
|
|
def as_dict(self) -> dict:
|
|
return {
|
|
"path": self.path,
|
|
"description": self.description,
|
|
"ok": self.ok,
|
|
"missing": self.missing,
|
|
"checks": [c.as_dict() for c in self.checks],
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class BaselineReport:
|
|
profile: str
|
|
ok: bool
|
|
variables: dict[str, str]
|
|
files: list[BaselineFileResult]
|
|
summary: str
|
|
|
|
def as_dict(self) -> dict:
|
|
return {
|
|
"profile": self.profile,
|
|
"ok": self.ok,
|
|
"variables": self.variables,
|
|
"files": [f.as_dict() for f in self.files],
|
|
"summary": self.summary,
|
|
}
|
|
|
|
|
|
def templates_root() -> str:
|
|
return os.environ.get("ASTERISK_TEMPLATES_PATH", settings.asterisk_templates_path)
|
|
|
|
|
|
def manifest_path(profile: str = "geneseasx") -> str:
|
|
return os.path.join(templates_root(), profile, "manifest.yaml")
|
|
|
|
|
|
def profile_for_catalog(catalog_key: str | None) -> str | None:
|
|
root = templates_root()
|
|
if not os.path.isdir(root):
|
|
return None
|
|
key = (catalog_key or "").lower()
|
|
for name in sorted(os.listdir(root)):
|
|
path = os.path.join(root, name, "manifest.yaml")
|
|
if not os.path.isfile(path):
|
|
continue
|
|
try:
|
|
with open(path, encoding="utf-8") as fh:
|
|
data = yaml.safe_load(fh) or {}
|
|
except (OSError, yaml.YAMLError):
|
|
continue
|
|
keys = [str(k).lower() for k in (data.get("catalog_keys") or [])]
|
|
if key in keys:
|
|
return name
|
|
return None
|
|
|
|
|
|
def load_manifest(profile: str, *, force: bool = False) -> dict | None:
|
|
global _manifest_cache, _manifest_mtime
|
|
path = manifest_path(profile)
|
|
try:
|
|
mtime = os.path.getmtime(path)
|
|
except OSError:
|
|
return None
|
|
|
|
if not force and _manifest_cache.get(profile) and _manifest_mtime == mtime:
|
|
return _manifest_cache[profile]
|
|
|
|
try:
|
|
with open(path, encoding="utf-8") as fh:
|
|
data = yaml.safe_load(fh) or {}
|
|
except (OSError, yaml.YAMLError) as exc:
|
|
logger.warning("invalid asterisk manifest %s: %s", path, exc)
|
|
return None
|
|
|
|
_manifest_cache[profile] = data
|
|
_manifest_mtime = mtime
|
|
return data
|
|
|
|
|
|
def invalidate_manifest_cache() -> None:
|
|
global _manifest_cache, _manifest_mtime
|
|
_manifest_cache = {}
|
|
_manifest_mtime = -1.0
|
|
|
|
|
|
def substitute(text: str, variables: dict[str, str]) -> str:
|
|
out = text
|
|
for name, value in variables.items():
|
|
out = out.replace(f"{{{{{name}}}}}", value)
|
|
out = out.replace(f"__{name}__", value)
|
|
return out
|
|
|
|
|
|
def resolve_variables(manifest: dict, device_ctx: dict | None = None) -> dict[str, str]:
|
|
"""Merge manifest defaults, env overrides, and optional device context."""
|
|
device_ctx = device_ctx or {}
|
|
variables: dict[str, str] = {
|
|
str(k): str(v) for k, v in (manifest.get("variables") or {}).items()
|
|
}
|
|
|
|
for name in list(variables.keys()):
|
|
env_key = f"ASTERISK_BASELINE_{name}"
|
|
env_val = os.environ.get(env_key, "").strip()
|
|
if env_val:
|
|
variables[name] = env_val
|
|
|
|
# Optional per-device overrides (inventory metadata or env-derived)
|
|
mapping = {
|
|
"ASTERISK_IP": device_ctx.get("asterisk_ip"),
|
|
"PHONE_SUBNET": device_ctx.get("phone_subnet"),
|
|
"VOIP_SUBNET": device_ctx.get("voip_subnet"),
|
|
"RTP_START": device_ctx.get("rtp_start"),
|
|
"RTP_END": device_ctx.get("rtp_end"),
|
|
}
|
|
for name, val in mapping.items():
|
|
if val:
|
|
variables[name] = str(val)
|
|
|
|
return variables
|
|
|
|
|
|
def _normalize_ini_line(line: str) -> str:
|
|
line = line.strip()
|
|
if not line or line.startswith(";") or line.startswith("#"):
|
|
return ""
|
|
line = line.split(";", 1)[0].strip()
|
|
line = re.sub(r"\s+", "", line)
|
|
return line.lower()
|
|
|
|
|
|
def _content_normalized_lines(content: str) -> list[str]:
|
|
return [n for n in (_normalize_ini_line(ln) for ln in content.splitlines()) if n]
|
|
|
|
|
|
def _line_present(content: str, required: str) -> bool:
|
|
req = _normalize_ini_line(required)
|
|
if not req:
|
|
return True
|
|
norm_lines = _content_normalized_lines(content)
|
|
if req in norm_lines:
|
|
return True
|
|
# Allow key=value match when live file has extra whitespace/casing
|
|
if "=" in req:
|
|
key, _, val = req.partition("=")
|
|
for line in norm_lines:
|
|
if line.startswith(key + "=") and val in line:
|
|
return True
|
|
return False
|
|
|
|
|
|
def _substring_present(content: str, needle: str) -> bool:
|
|
return needle.strip().lower() in content.lower()
|
|
|
|
|
|
def compare_file(
|
|
live_content: str | None,
|
|
file_spec: dict,
|
|
variables: dict[str, str],
|
|
) -> BaselineFileResult:
|
|
path = str(file_spec.get("path") or "")
|
|
description = str(file_spec.get("description") or "")
|
|
checks: list[BaselineCheck] = []
|
|
|
|
if live_content is None or live_content.strip() == "__MISSING__":
|
|
return BaselineFileResult(
|
|
path=path,
|
|
description=description,
|
|
ok=False,
|
|
missing=True,
|
|
checks=checks,
|
|
)
|
|
|
|
for raw in file_spec.get("required_lines") or []:
|
|
expected = substitute(str(raw), variables)
|
|
found = _line_present(live_content, expected)
|
|
checks.append(
|
|
BaselineCheck(
|
|
kind="required_line",
|
|
expected=expected,
|
|
found=found,
|
|
ok=found,
|
|
)
|
|
)
|
|
|
|
for raw in file_spec.get("required_substrings") or []:
|
|
expected = substitute(str(raw), variables)
|
|
found = _substring_present(live_content, expected)
|
|
checks.append(
|
|
BaselineCheck(
|
|
kind="required_substring",
|
|
expected=expected,
|
|
found=found,
|
|
ok=found,
|
|
)
|
|
)
|
|
|
|
ok = all(c.ok for c in checks) if checks else True
|
|
return BaselineFileResult(path=path, description=description, ok=ok, checks=checks)
|
|
|
|
|
|
def compare_against_baseline(
|
|
live_files: dict[str, str],
|
|
*,
|
|
profile: str,
|
|
device_ctx: dict | None = None,
|
|
) -> BaselineReport | None:
|
|
manifest = load_manifest(profile)
|
|
if not manifest:
|
|
return None
|
|
|
|
variables = resolve_variables(manifest, device_ctx)
|
|
file_results: list[BaselineFileResult] = []
|
|
|
|
for file_spec in manifest.get("files") or []:
|
|
path = str(file_spec.get("path") or "")
|
|
live = live_files.get(path)
|
|
file_results.append(compare_file(live, file_spec, variables))
|
|
|
|
total_checks = sum(len(f.checks) for f in file_results)
|
|
passed = sum(1 for f in file_results for c in f.checks if c.ok)
|
|
files_ok = all(f.ok for f in file_results) if file_results else True
|
|
|
|
if any(f.missing for f in file_results):
|
|
summary = f"Config baseline: {passed}/{total_checks} checks passed; some files missing"
|
|
elif total_checks == 0:
|
|
summary = "Config baseline: no checks defined"
|
|
else:
|
|
summary = f"Config baseline: {passed}/{total_checks} checks passed"
|
|
|
|
return BaselineReport(
|
|
profile=profile,
|
|
ok=files_ok and passed == total_checks,
|
|
variables=variables,
|
|
files=file_results,
|
|
summary=summary,
|
|
)
|
|
|
|
|
|
def config_paths_for_profile(profile: str) -> list[str]:
|
|
manifest = load_manifest(profile)
|
|
if not manifest:
|
|
return []
|
|
return [str(f.get("path")) for f in (manifest.get("files") or []) if f.get("path")]
|
|
|
|
|
|
def config_root_for_profile(profile: str) -> str:
|
|
manifest = load_manifest(profile) or {}
|
|
return str(manifest.get("config_root") or "/etc/asterisk").rstrip("/")
|