88 lines
2.7 KiB
Python
88 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Utilities for discovering and parsing Nginx configuration files.
|
|
|
|
This module provides helper functions to locate Nginx configuration files and
|
|
extract key details from ``server`` blocks. Typical usage::
|
|
|
|
from scripts.nginx_config import discover_configs, parse_servers
|
|
|
|
files = discover_configs()
|
|
servers = parse_servers(files)
|
|
for s in servers:
|
|
print(s.get("server_name"), s.get("listen"))
|
|
|
|
The functions intentionally tolerate missing or unreadable files and will simply
|
|
skip over them.
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Dict, List, Set
|
|
|
|
DEFAULT_PATHS = [
|
|
"/etc/nginx/nginx.conf",
|
|
"/usr/local/etc/nginx/nginx.conf",
|
|
]
|
|
|
|
INCLUDE_RE = re.compile(r"^\s*include\s+(.*?);", re.MULTILINE)
|
|
SERVER_RE = re.compile(r"server\s*{(.*?)}", re.DOTALL)
|
|
DIRECTIVE_RE = re.compile(r"^\s*(\S+)\s+(.*?);", re.MULTILINE)
|
|
|
|
|
|
def discover_configs() -> Set[Path]:
|
|
"""Return a set of all config files reachable from :data:`DEFAULT_PATHS`."""
|
|
|
|
found: Set[Path] = set()
|
|
queue = [Path(p) for p in DEFAULT_PATHS]
|
|
|
|
while queue:
|
|
path = queue.pop()
|
|
if path in found:
|
|
continue
|
|
if not path.exists():
|
|
continue
|
|
try:
|
|
text = path.read_text()
|
|
except OSError:
|
|
continue
|
|
found.add(path)
|
|
for pattern in INCLUDE_RE.findall(text):
|
|
pattern = os.path.expanduser(pattern.strip())
|
|
for included in path.parent.glob(pattern):
|
|
if included.is_file() and included not in found:
|
|
queue.append(included)
|
|
return found
|
|
|
|
|
|
def parse_servers(paths: Set[Path]) -> List[Dict[str, str]]:
|
|
"""Parse ``server`` blocks from the given files.
|
|
|
|
Parameters
|
|
----------
|
|
paths:
|
|
Iterable of configuration file paths.
|
|
"""
|
|
|
|
servers: List[Dict[str, str]] = []
|
|
for p in paths:
|
|
try:
|
|
text = Path(p).read_text()
|
|
except OSError:
|
|
continue
|
|
for block in SERVER_RE.findall(text):
|
|
directives: Dict[str, List[str]] = {}
|
|
for name, value in DIRECTIVE_RE.findall(block):
|
|
directives.setdefault(name, []).append(value.strip())
|
|
entry: Dict[str, str] = {}
|
|
if "server_name" in directives:
|
|
entry["server_name"] = " ".join(directives["server_name"])
|
|
if "listen" in directives:
|
|
entry["listen"] = " ".join(directives["listen"])
|
|
if "proxy_cache" in directives:
|
|
entry["proxy_cache"] = " ".join(directives["proxy_cache"])
|
|
if "root" in directives:
|
|
entry["root"] = " ".join(directives["root"])
|
|
servers.append(entry)
|
|
return servers
|
|
|