homelab: sync post-migration repo and n8n runtime audit
This commit is contained in:
Fizzlepoof
134
automation/bin/unifi_stage_firewall_groups.py
Normal file
134
automation/bin/unifi_stage_firewall_groups.py
Normal file
@@ -0,0 +1,134 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Stage UniFi firewall groups for the current network segmentation.
|
||||
|
||||
Default mode is dry-run. Pass --apply to create/update/delete as needed.
|
||||
Reads credentials from the runtime automation .env on PD by default.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
DEFAULT_RUNTIME_HELPER = Path('/mnt/docker-ssd/docker/compose/automation/bin/unifi_stage_low_risk_objects.py')
|
||||
DEFAULT_ENV_PATH = Path('/mnt/docker-ssd/docker/compose/automation/.env')
|
||||
|
||||
|
||||
def load_helper(path: Path):
|
||||
spec = importlib.util.spec_from_file_location('unifi_stage', path)
|
||||
if spec is None or spec.loader is None:
|
||||
raise RuntimeError(f'Could not load helper module from {path}')
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def desired_groups() -> list[dict[str, Any]]:
|
||||
return [
|
||||
{'name': 'NET-MGMT', 'group_type': 'address-group', 'group_members': ['10.5.0.0/24']},
|
||||
{'name': 'NET-TRUSTED', 'group_type': 'address-group', 'group_members': ['10.5.1.0/24']},
|
||||
{'name': 'NET-SERVERS', 'group_type': 'address-group', 'group_members': ['10.5.30.0/24']},
|
||||
{'name': 'NET-IOT', 'group_type': 'address-group', 'group_members': ['10.5.10.0/24']},
|
||||
{'name': 'NET-GUEST', 'group_type': 'address-group', 'group_members': ['10.5.90.0/24']},
|
||||
{'name': 'NET-CAMERAS', 'group_type': 'address-group', 'group_members': ['10.5.20.0/24']},
|
||||
{'name': 'NET-LEGACY-CIA', 'group_type': 'address-group', 'group_members': ['192.168.1.0/24']},
|
||||
{
|
||||
'name': 'NET-RFC1918-ALL',
|
||||
'group_type': 'address-group',
|
||||
'group_members': [
|
||||
'10.5.0.0/24',
|
||||
'10.5.1.0/24',
|
||||
'10.5.10.0/24',
|
||||
'10.5.20.0/24',
|
||||
'10.5.30.0/24',
|
||||
'10.5.90.0/24',
|
||||
'192.168.1.0/24',
|
||||
'192.168.2.0/24',
|
||||
'192.168.3.0/24',
|
||||
],
|
||||
},
|
||||
{'name': 'PORT-DNS', 'group_type': 'port-group', 'group_members': ['53']},
|
||||
{'name': 'PORT-DHCP', 'group_type': 'port-group', 'group_members': ['67-68']},
|
||||
{'name': 'PORT-NTP', 'group_type': 'port-group', 'group_members': ['123']},
|
||||
{'name': 'PORT-WEB-ADMIN', 'group_type': 'port-group', 'group_members': ['80', '443', '8443', '9443']},
|
||||
{'name': 'PORT-SSH', 'group_type': 'port-group', 'group_members': ['22']},
|
||||
{'name': 'PORT-MDNS', 'group_type': 'port-group', 'group_members': ['5353']},
|
||||
]
|
||||
|
||||
|
||||
TEMP_GROUP_NAMES = {'TEST-TMP', 'TEST-PORT-TMP', 'TEST-PORT-TMP2', 'TEST-PORT-TMP3'}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description='Stage UniFi firewall groups.')
|
||||
parser.add_argument('--helper', default=str(DEFAULT_RUNTIME_HELPER), help='Path to runtime UniFi helper on PD')
|
||||
parser.add_argument('--env-file', default=str(DEFAULT_ENV_PATH), help='Path to runtime .env on PD')
|
||||
parser.add_argument('--site', default='default')
|
||||
parser.add_argument('--apply', action='store_true')
|
||||
args = parser.parse_args()
|
||||
|
||||
helper = load_helper(Path(args.helper))
|
||||
helper.load_env_file(Path(args.env_file))
|
||||
client = helper.UniFiClient(
|
||||
base_url=os.environ.get('UNIFI_BASE_URL', 'https://10.5.0.1'),
|
||||
username=os.environ['UNIFI_USERNAME'],
|
||||
password=os.environ['UNIFI_PASSWORD'],
|
||||
verify_tls=helper.env_bool('UNIFI_VERIFY_TLS', False),
|
||||
timeout=20,
|
||||
)
|
||||
client.login()
|
||||
|
||||
groups = client.get(client.site_path(args.site, 'rest/firewallgroup'))
|
||||
by_name = {g.get('name'): g for g in groups}
|
||||
result: dict[str, Any] = {'mode': 'apply' if args.apply else 'dry-run', 'changes': []}
|
||||
|
||||
for group in desired_groups():
|
||||
existing = by_name.get(group['name'])
|
||||
if not existing:
|
||||
result['changes'].append({'name': group['name'], 'action': 'create', 'payload': group})
|
||||
if args.apply:
|
||||
status, payload = client._json_request('POST', client.site_path(args.site, 'rest/firewallgroup'), group)
|
||||
if status >= 300:
|
||||
result['changes'][-1]['error'] = {'status': status, 'payload': payload}
|
||||
continue
|
||||
|
||||
current_members = existing.get('group_members', []) or []
|
||||
desired_members = group.get('group_members', []) or []
|
||||
current_type = existing.get('group_type')
|
||||
if current_type != group['group_type'] or current_members != desired_members:
|
||||
updated = dict(existing)
|
||||
updated['group_type'] = group['group_type']
|
||||
updated['group_members'] = desired_members
|
||||
result['changes'].append({
|
||||
'name': group['name'],
|
||||
'action': 'update',
|
||||
'before': {'group_type': current_type, 'group_members': current_members},
|
||||
'after': {'group_type': group['group_type'], 'group_members': desired_members},
|
||||
})
|
||||
if args.apply:
|
||||
status, payload = client._json_request('PUT', client.site_path(args.site, f"rest/firewallgroup/{existing['_id']}"), updated)
|
||||
if status >= 300:
|
||||
result['changes'][-1]['error'] = {'status': status, 'payload': payload}
|
||||
else:
|
||||
result['changes'].append({'name': group['name'], 'action': 'noop'})
|
||||
|
||||
for name in sorted(TEMP_GROUP_NAMES):
|
||||
existing = by_name.get(name)
|
||||
if not existing:
|
||||
continue
|
||||
result['changes'].append({'name': name, 'action': 'delete-temp'})
|
||||
if args.apply:
|
||||
status, payload = client._json_request('DELETE', client.site_path(args.site, f"rest/firewallgroup/{existing['_id']}"))
|
||||
if status >= 300:
|
||||
result['changes'][-1]['error'] = {'status': status, 'payload': payload}
|
||||
|
||||
print(json.dumps(result, indent=2, sort_keys=True))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
raise SystemExit(main())
|
||||
201
automation/bin/unifi_stage_policy_engine_firewall.py
Normal file
201
automation/bin/unifi_stage_policy_engine_firewall.py
Normal file
@@ -0,0 +1,201 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Stage UniFi Policy Engine firewall policies for the current segmentation.
|
||||
|
||||
Default mode is dry-run. Pass --apply to create/update desired custom policies.
|
||||
Reads credentials from the runtime automation .env on PD by default.
|
||||
|
||||
This helper targets the newer Policy Engine / zone-based firewall model via:
|
||||
/proxy/network/v2/api/site/<site>/firewall/zone
|
||||
/proxy/network/v2/api/site/<site>/firewall-policies
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
DEFAULT_RUNTIME_HELPER = Path('/mnt/docker-ssd/docker/compose/automation/bin/unifi_stage_low_risk_objects.py')
|
||||
DEFAULT_ENV_PATH = Path('/mnt/docker-ssd/docker/compose/automation/.env')
|
||||
|
||||
|
||||
def load_helper(path: Path):
|
||||
spec = importlib.util.spec_from_file_location('unifi_stage', path)
|
||||
if spec is None or spec.loader is None:
|
||||
raise RuntimeError(f'Could not load helper module from {path}')
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def desired_policies() -> list[dict[str, Any]]:
|
||||
return [
|
||||
{
|
||||
'name': 'Allow Internal to Untrusted',
|
||||
'enabled': True,
|
||||
'action': 'ALLOW',
|
||||
'protocol': 'all',
|
||||
'ip_version': 'BOTH',
|
||||
'logging': False,
|
||||
'create_allow_respond': True,
|
||||
'connection_state_type': 'ALL',
|
||||
'connection_states': [],
|
||||
'match_ip_sec': False,
|
||||
'match_opposite_protocol': False,
|
||||
'icmp_typename': 'ANY',
|
||||
'icmp_v6_typename': 'ANY',
|
||||
'source_zone_name': 'Internal',
|
||||
'destination_zone_name': 'Untrusted',
|
||||
'schedule': {'mode': 'ALWAYS'},
|
||||
'source': {
|
||||
'matching_target': 'ANY',
|
||||
'port_matching_type': 'ANY',
|
||||
'match_opposite_ports': False,
|
||||
},
|
||||
'destination': {
|
||||
'matching_target': 'ANY',
|
||||
'port_matching_type': 'ANY',
|
||||
'match_opposite_ports': False,
|
||||
},
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
POLICY_MUTABLE_KEYS = {
|
||||
'name',
|
||||
'enabled',
|
||||
'action',
|
||||
'protocol',
|
||||
'ip_version',
|
||||
'logging',
|
||||
'create_allow_respond',
|
||||
'connection_state_type',
|
||||
'connection_states',
|
||||
'match_ip_sec',
|
||||
'match_opposite_protocol',
|
||||
'icmp_typename',
|
||||
'icmp_v6_typename',
|
||||
'source',
|
||||
'destination',
|
||||
'schedule',
|
||||
'description',
|
||||
}
|
||||
|
||||
|
||||
def canonical_policy(policy: dict[str, Any]) -> dict[str, Any]:
|
||||
out = {key: deepcopy(policy.get(key)) for key in POLICY_MUTABLE_KEYS if key in policy}
|
||||
out.setdefault('description', '')
|
||||
out.setdefault('connection_state_type', 'ALL')
|
||||
out.setdefault('connection_states', [])
|
||||
out.setdefault('logging', False)
|
||||
out.setdefault('match_ip_sec', False)
|
||||
out.setdefault('match_opposite_protocol', False)
|
||||
out.setdefault('icmp_typename', 'ANY')
|
||||
out.setdefault('icmp_v6_typename', 'ANY')
|
||||
out.setdefault('schedule', {'mode': 'ALWAYS'})
|
||||
return out
|
||||
|
||||
|
||||
def resolve_zone_ids(zone_data: list[dict[str, Any]]) -> dict[str, str]:
|
||||
by_name: dict[str, str] = {}
|
||||
for zone in zone_data:
|
||||
name = zone.get('name')
|
||||
zid = zone.get('_id')
|
||||
if name and zid:
|
||||
by_name[name] = zid
|
||||
return by_name
|
||||
|
||||
|
||||
def materialize_policy(spec: dict[str, Any], zone_ids: dict[str, str]) -> dict[str, Any]:
|
||||
source_zone = spec['source_zone_name']
|
||||
dest_zone = spec['destination_zone_name']
|
||||
if source_zone not in zone_ids:
|
||||
raise KeyError(f'Missing source zone: {source_zone}')
|
||||
if dest_zone not in zone_ids:
|
||||
raise KeyError(f'Missing destination zone: {dest_zone}')
|
||||
|
||||
policy = {k: deepcopy(v) for k, v in spec.items() if not k.endswith('_zone_name')}
|
||||
policy['description'] = policy.get('description', '')
|
||||
policy['source'] = deepcopy(policy['source'])
|
||||
policy['destination'] = deepcopy(policy['destination'])
|
||||
policy['source']['zone_id'] = zone_ids[source_zone]
|
||||
policy['destination']['zone_id'] = zone_ids[dest_zone]
|
||||
return policy
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description='Stage UniFi Policy Engine firewall policies.')
|
||||
parser.add_argument('--helper', default=str(DEFAULT_RUNTIME_HELPER), help='Path to runtime UniFi helper on PD')
|
||||
parser.add_argument('--env-file', default=str(DEFAULT_ENV_PATH), help='Path to runtime .env on PD')
|
||||
parser.add_argument('--site', default='default')
|
||||
parser.add_argument('--apply', action='store_true')
|
||||
args = parser.parse_args()
|
||||
|
||||
helper = load_helper(Path(args.helper))
|
||||
helper.load_env_file(Path(args.env_file))
|
||||
client = helper.UniFiClient(
|
||||
base_url=os.environ.get('UNIFI_BASE_URL', 'https://10.5.0.1'),
|
||||
username=os.environ['UNIFI_USERNAME'],
|
||||
password=os.environ['UNIFI_PASSWORD'],
|
||||
verify_tls=helper.env_bool('UNIFI_VERIFY_TLS', False),
|
||||
timeout=20,
|
||||
)
|
||||
client.login()
|
||||
|
||||
zones_payload = client.get(f'/proxy/network/v2/api/site/{args.site}/firewall/zone')
|
||||
zone_data_raw = zones_payload['data'] if isinstance(zones_payload, dict) and 'data' in zones_payload else zones_payload
|
||||
zone_data = zone_data_raw if isinstance(zone_data_raw, list) else []
|
||||
zone_ids = resolve_zone_ids(zone_data)
|
||||
|
||||
policies_path = f'/proxy/network/v2/api/site/{args.site}/firewall-policies'
|
||||
policies_payload = client.get(policies_path)
|
||||
policy_data_raw = policies_payload['data'] if isinstance(policies_payload, dict) and 'data' in policies_payload else policies_payload
|
||||
policy_data = policy_data_raw if isinstance(policy_data_raw, list) else []
|
||||
custom_by_name = {p.get('name'): p for p in policy_data if not p.get('predefined') and p.get('name')}
|
||||
|
||||
result: dict[str, Any] = {
|
||||
'mode': 'apply' if args.apply else 'dry-run',
|
||||
'zones': sorted(zone_ids.keys()),
|
||||
'changes': [],
|
||||
}
|
||||
|
||||
for desired in desired_policies():
|
||||
materialized = materialize_policy(desired, zone_ids)
|
||||
existing = custom_by_name.get(materialized['name'])
|
||||
desired_canonical = canonical_policy(materialized)
|
||||
if existing is None:
|
||||
result['changes'].append({'name': materialized['name'], 'action': 'create', 'payload': desired_canonical})
|
||||
if args.apply:
|
||||
status, payload = client._json_request('POST', policies_path, desired_canonical)
|
||||
if status >= 300:
|
||||
result['changes'][-1]['error'] = {'status': status, 'payload': payload}
|
||||
continue
|
||||
|
||||
existing_canonical = canonical_policy(existing)
|
||||
if existing_canonical != desired_canonical:
|
||||
update_payload = deepcopy(existing)
|
||||
for key, value in desired_canonical.items():
|
||||
update_payload[key] = deepcopy(value)
|
||||
result['changes'].append({
|
||||
'name': materialized['name'],
|
||||
'action': 'update',
|
||||
'before': existing_canonical,
|
||||
'after': desired_canonical,
|
||||
})
|
||||
if args.apply:
|
||||
status, payload = client._json_request('PUT', f"{policies_path}/{existing['_id']}", update_payload)
|
||||
if status >= 300:
|
||||
result['changes'][-1]['error'] = {'status': status, 'payload': payload}
|
||||
else:
|
||||
result['changes'].append({'name': materialized['name'], 'action': 'noop'})
|
||||
|
||||
print(json.dumps(result, indent=2, sort_keys=True))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user