Sync network redesign repo state and cutover artifacts
This commit is contained in:
Fizzlepoof
@@ -29,7 +29,7 @@ UNIFI_SITE=default
|
||||
UNIFI_VERIFY_TLS=false
|
||||
|
||||
# Backup automation scaffolding
|
||||
SERENITY_BACKUP_HOST=root@10.5.1.5
|
||||
SERENITY_BACKUP_HOST=root@10.5.30.5
|
||||
SERENITY_BACKUP_ROOT=/mnt/user/backups/plausible-deniability
|
||||
PD_APPDATA_ROOT=/mnt/docker-ssd/docker/appdata
|
||||
PD_DATABASES_ROOT=/mnt/docker-ssd/docker/databases
|
||||
|
||||
@@ -44,7 +44,7 @@ Repo-side deployment prerequisites remain:
|
||||
|
||||
1. real values in `.env`
|
||||
2. a writable dump destination on PD
|
||||
3. SSH trust / key path for Serenity (`root@10.5.1.5` by default)
|
||||
3. SSH trust / key path for Serenity (`root@10.5.30.5` by default)
|
||||
- recommended known_hosts path on PD: `/home/truenas_admin/.ssh/known_hosts`
|
||||
4. a real scheduler on the live host
|
||||
|
||||
|
||||
270
automation/bin/unifi_stage_low_risk_objects.py
Executable file
270
automation/bin/unifi_stage_low_risk_objects.py
Executable file
@@ -0,0 +1,270 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Stage low-risk UniFi objects for the network cutover.
|
||||
|
||||
Default mode is read-only dry-run. Pass --apply to actually create missing objects.
|
||||
Currently stages:
|
||||
- Servers network object
|
||||
- Servers port profile
|
||||
|
||||
This is intentionally idempotent: if the objects already exist, it reports them and exits cleanly.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import ssl
|
||||
import sys
|
||||
from http.cookiejar import CookieJar
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib import error, request
|
||||
|
||||
DEFAULT_TIMEOUT = 20
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
REPO_ROOT = SCRIPT_DIR.parent.parent
|
||||
DEFAULT_ENV_PATH = REPO_ROOT / 'automation' / '.env'
|
||||
|
||||
|
||||
def load_env_file(path: Path) -> None:
|
||||
if not path.exists():
|
||||
return
|
||||
for raw_line in path.read_text(encoding='utf-8').splitlines():
|
||||
line = raw_line.strip()
|
||||
if not line or line.startswith('#') or '=' not in line:
|
||||
continue
|
||||
key, value = line.split('=', 1)
|
||||
os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'"))
|
||||
|
||||
|
||||
def env_bool(name: str, default: bool) -> bool:
|
||||
value = os.environ.get(name)
|
||||
if value is None:
|
||||
return default
|
||||
return value.strip().lower() in {'1', 'true', 'yes', 'on'}
|
||||
|
||||
|
||||
class UniFiClient:
|
||||
def __init__(self, base_url: str, username: str, password: str, *, verify_tls: bool = True, timeout: int = DEFAULT_TIMEOUT) -> None:
|
||||
self.base_url = base_url.rstrip('/')
|
||||
self.username = username
|
||||
self.password = password
|
||||
self.timeout = timeout
|
||||
self.cookie_jar = CookieJar()
|
||||
self.csrf_token: str | None = None
|
||||
handlers: list[Any] = [request.HTTPCookieProcessor(self.cookie_jar)]
|
||||
if not verify_tls:
|
||||
handlers.append(request.HTTPSHandler(context=ssl._create_unverified_context()))
|
||||
self.opener = request.build_opener(*handlers)
|
||||
|
||||
def _json_request(self, method: str, path: str, body: dict[str, Any] | None = None) -> tuple[int, Any]:
|
||||
url = self.base_url + path
|
||||
data = None
|
||||
headers = {
|
||||
'Accept': 'application/json',
|
||||
'Content-Type': 'application/json',
|
||||
'User-Agent': 'doris-unifi-stage/1.0',
|
||||
}
|
||||
if method.upper() != 'GET' and self.csrf_token:
|
||||
headers['X-Csrf-Token'] = self.csrf_token
|
||||
if body is not None:
|
||||
data = json.dumps(body).encode('utf-8')
|
||||
req = request.Request(url, data=data, method=method.upper(), headers=headers)
|
||||
try:
|
||||
with self.opener.open(req, timeout=self.timeout) as resp:
|
||||
if method.upper() == 'POST':
|
||||
self.csrf_token = resp.headers.get('X-Csrf-Token', self.csrf_token)
|
||||
raw = resp.read().decode('utf-8', 'replace')
|
||||
ctype = resp.headers.get('Content-Type', '')
|
||||
payload = json.loads(raw) if 'json' in ctype and raw else raw
|
||||
return resp.status, payload
|
||||
except error.HTTPError as exc:
|
||||
raw = exc.read().decode('utf-8', 'replace')
|
||||
try:
|
||||
payload = json.loads(raw) if raw else {}
|
||||
except Exception:
|
||||
payload = raw
|
||||
return exc.code, payload
|
||||
|
||||
def login(self) -> None:
|
||||
status, payload = self._json_request('POST', '/api/auth/login', {
|
||||
'username': self.username,
|
||||
'password': self.password,
|
||||
'rememberMe': True,
|
||||
})
|
||||
if status >= 300:
|
||||
raise SystemExit(f'UniFi login failed: HTTP {status} {payload!r}')
|
||||
|
||||
def get(self, path: str) -> Any:
|
||||
status, payload = self._json_request('GET', path)
|
||||
if status >= 300:
|
||||
raise SystemExit(f'UniFi GET {path} failed: HTTP {status} {payload!r}')
|
||||
return payload.get('data', payload) if isinstance(payload, dict) else payload
|
||||
|
||||
def post(self, path: str, body: dict[str, Any]) -> Any:
|
||||
status, payload = self._json_request('POST', path, body)
|
||||
if status >= 300:
|
||||
raise SystemExit(f'UniFi POST {path} failed: HTTP {status} {payload!r}')
|
||||
return payload.get('data', payload) if isinstance(payload, dict) else payload
|
||||
|
||||
def site_path(self, site: str, suffix: str) -> str:
|
||||
suffix = suffix.lstrip('/')
|
||||
return f'/proxy/network/api/s/{site}/{suffix}'
|
||||
|
||||
|
||||
def find_by_name(items: list[dict[str, Any]], name: str) -> dict[str, Any] | None:
|
||||
return next((item for item in items if item.get('name') == name), None)
|
||||
|
||||
|
||||
def servers_network_payload() -> dict[str, Any]:
|
||||
return {
|
||||
'name': 'Servers',
|
||||
'purpose': 'corporate',
|
||||
'enabled': True,
|
||||
'networkgroup': 'LAN',
|
||||
'setting_preference': 'manual',
|
||||
'vlan_enabled': True,
|
||||
'vlan': 30,
|
||||
'ip_subnet': '10.5.30.1/24',
|
||||
'dhcpd_enabled': True,
|
||||
'dhcpd_start': '10.5.30.100',
|
||||
'dhcpd_stop': '10.5.30.199',
|
||||
'dhcpd_leasetime': 86400,
|
||||
'dhcpd_gateway_enabled': False,
|
||||
'dhcpd_dns_enabled': False,
|
||||
'dhcpd_ntp_enabled': False,
|
||||
'dhcpd_wins_enabled': False,
|
||||
'dhcpd_time_offset_enabled': False,
|
||||
'dhcpd_boot_enabled': False,
|
||||
'dhcpd_conflict_checking': True,
|
||||
'dhcp_relay_enabled': False,
|
||||
'dhcp_relay_servers': [],
|
||||
'domain_name': '',
|
||||
'ipv6_interface_type': 'none',
|
||||
'ipv6_setting_preference': 'manual',
|
||||
'ipv6_aliases': [],
|
||||
'ip_aliases': [],
|
||||
'internet_access_enabled': True,
|
||||
'network_isolation_enabled': False,
|
||||
'gateway_type': 'default',
|
||||
'nat_outbound_ip_addresses': [],
|
||||
'mdns_enabled': True,
|
||||
'lte_lan_enabled': True,
|
||||
'auto_scale_enabled': True,
|
||||
}
|
||||
|
||||
|
||||
def build_servers_port_profile_payload(template: dict[str, Any], servers_network_id: str) -> dict[str, Any]:
|
||||
payload = {
|
||||
k: v for k, v in template.items()
|
||||
if k not in {'_id', 'site_id', 'name', 'native_networkconf_id'}
|
||||
}
|
||||
payload['name'] = 'Servers'
|
||||
payload['native_networkconf_id'] = servers_network_id
|
||||
return payload
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description='Stage low-risk UniFi objects for the cutover.')
|
||||
parser.add_argument('--env-file', default=str(DEFAULT_ENV_PATH), help='Path to .env containing UNIFI_* vars')
|
||||
parser.add_argument('--site', help='UniFi site name; defaults to UNIFI_SITE or default')
|
||||
parser.add_argument('--apply', action='store_true', help='Actually create missing objects. Default is dry-run only.')
|
||||
parser.add_argument('--timeout', type=int, default=DEFAULT_TIMEOUT)
|
||||
args = parser.parse_args()
|
||||
|
||||
load_env_file(Path(args.env_file))
|
||||
base_url = os.environ.get('UNIFI_BASE_URL', 'https://10.5.0.1')
|
||||
username = os.environ.get('UNIFI_USERNAME')
|
||||
password = os.environ.get('UNIFI_PASSWORD')
|
||||
site = args.site or os.environ.get('UNIFI_SITE', 'default')
|
||||
verify_tls = env_bool('UNIFI_VERIFY_TLS', False)
|
||||
|
||||
if not username or not password:
|
||||
raise SystemExit('Missing UNIFI_USERNAME or UNIFI_PASSWORD in env file/environment.')
|
||||
|
||||
client = UniFiClient(base_url, username, password, verify_tls=verify_tls, timeout=args.timeout)
|
||||
client.login()
|
||||
|
||||
networks = client.get(client.site_path(site, 'rest/networkconf'))
|
||||
port_profiles = client.get(client.site_path(site, 'rest/portconf'))
|
||||
|
||||
result: dict[str, Any] = {
|
||||
'mode': 'apply' if args.apply else 'dry-run',
|
||||
'site': site,
|
||||
'changes': [],
|
||||
}
|
||||
|
||||
servers_network = find_by_name(networks, 'Servers')
|
||||
if servers_network:
|
||||
result['changes'].append({
|
||||
'object': 'network',
|
||||
'name': 'Servers',
|
||||
'status': 'exists',
|
||||
'id': servers_network.get('_id'),
|
||||
'vlan': servers_network.get('vlan'),
|
||||
'ip_subnet': servers_network.get('ip_subnet'),
|
||||
})
|
||||
else:
|
||||
payload = servers_network_payload()
|
||||
if args.apply:
|
||||
created = client.post(client.site_path(site, 'rest/networkconf'), payload)
|
||||
if isinstance(created, list):
|
||||
created = created[0] if created else {}
|
||||
servers_network = created
|
||||
result['changes'].append({
|
||||
'object': 'network',
|
||||
'name': 'Servers',
|
||||
'status': 'created',
|
||||
'id': created.get('_id'),
|
||||
'vlan': created.get('vlan'),
|
||||
'ip_subnet': created.get('ip_subnet'),
|
||||
})
|
||||
else:
|
||||
servers_network = {'_id': '<created-network-id>'}
|
||||
result['changes'].append({
|
||||
'object': 'network',
|
||||
'name': 'Servers',
|
||||
'status': 'would_create',
|
||||
'payload': payload,
|
||||
})
|
||||
|
||||
servers_port_profile = find_by_name(port_profiles, 'Servers')
|
||||
if servers_port_profile:
|
||||
result['changes'].append({
|
||||
'object': 'port_profile',
|
||||
'name': 'Servers',
|
||||
'status': 'exists',
|
||||
'id': servers_port_profile.get('_id'),
|
||||
'native_networkconf_id': servers_port_profile.get('native_networkconf_id'),
|
||||
})
|
||||
else:
|
||||
trusted_template = find_by_name(port_profiles, 'Trusted')
|
||||
if not trusted_template:
|
||||
raise SystemExit('Could not find existing Trusted port profile to clone.')
|
||||
payload = build_servers_port_profile_payload(trusted_template, servers_network['_id'])
|
||||
if args.apply:
|
||||
created = client.post(client.site_path(site, 'rest/portconf'), payload)
|
||||
if isinstance(created, list):
|
||||
created = created[0] if created else {}
|
||||
result['changes'].append({
|
||||
'object': 'port_profile',
|
||||
'name': 'Servers',
|
||||
'status': 'created',
|
||||
'id': created.get('_id'),
|
||||
'native_networkconf_id': created.get('native_networkconf_id'),
|
||||
})
|
||||
else:
|
||||
result['changes'].append({
|
||||
'object': 'port_profile',
|
||||
'name': 'Servers',
|
||||
'status': 'would_create',
|
||||
'payload': payload,
|
||||
})
|
||||
|
||||
print(json.dumps(result, indent=2, sort_keys=True))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user