#!/usr/bin/env python3 """Track narrowly-scoped SRS virtual alias additions in a Postfix main.cf. This helper never restores an old main.cf; it only adds/removes its exact two list items. An ownership marker is mandatory before removal. This is a review draft: do not use on a live mail server until the complete integration and rollback have been tested. """ import argparse import os import re import stat import sys import tempfile from pathlib import Path MARKER_PREFIX = '# OOPEN-SRS-ALIAS-OWNERSHIP v1 ' PARAMS = ('virtual_alias_domains', 'virtual_alias_maps') class ConfigError(Exception): pass def read_entries(lines, param): """Return [(first_line, last_line_exclusive, value_lines)] for a parameter. Continuation lines must be indented, active, and not comments. A comment ends the immediately preceding value, leaving its original text untouched. """ pattern = re.compile(r'^\s*' + re.escape(param) + r'\s*=', re.I) found = [] for i, line in enumerate(lines): if not pattern.match(line) or line.lstrip().startswith('#'): continue j = i + 1 while j < len(lines) and lines[j].startswith((' ', '\t')) and lines[j].strip() and not lines[j].lstrip().startswith('#'): j += 1 first_value = line.split('=', 1)[1].strip() values = [first_value] + [item.strip() for item in lines[i + 1:j]] found.append((i, j, values)) if len(found) > 1: raise ConfigError(f'Duplicate active parameter {param}') return found def current_tokens(lines, param): entries = read_entries(lines, param) if not entries: return (False, []) values = entries[0][2] return (True, [part for part in re.split(r'[\s,]+', ' '.join(values)) if part]) def make_marker(domain, reverse_map, orig): return (MARKER_PREFIX + f'domain={domain} map={reverse_map} ' f'domains_was_set={int(orig[0])} maps_was_set={int(orig[1])}\n') def marker_state(lines, domain, reverse_map): markers = [line for line in lines if line.startswith('# OOPEN-SRS-ALIAS-OWNERSHIP')] if not markers: return None if len(markers) != 1: raise ConfigError('Multiple SRS alias ownership markers') marker = markers[0] expected = re.compile(r'^' + re.escape(MARKER_PREFIX) + r'domain=' + re.escape(domain) + r' map=' + re.escape(reverse_map) + r' domains_was_set=([01]) maps_was_set=([01])\n$') match = expected.fullmatch(marker) if not match: raise ConfigError('Foreign or malformed SRS alias ownership marker') return tuple(item == '1' for item in match.groups()) def inspect(lines, domain, reverse_map): is_set_domains, domains = current_tokens(lines, PARAMS[0]) is_set_maps, maps = current_tokens(lines, PARAMS[1]) matches_domain = sum(token.lower() == domain.lower() for token in domains) matches_map = sum(token == reverse_map for token in maps) if matches_domain > 1 or matches_map > 1: raise ConfigError('Duplicate SRS domain or reverse map item') ownership = marker_state(lines, domain, reverse_map) if ownership is not None: if matches_domain != 1 or matches_map != 1: raise ConfigError('Ownership marker present but SRS tokens are incomplete') return 'managed', ownership if matches_domain and matches_map: return 'legacy', None if not matches_domain and not matches_map: return 'absent', (is_set_domains, is_set_maps) raise ConfigError('Partial/unmanaged SRS alias configuration') def append_token(lines, param, token): entries = read_entries(lines, param) if not entries: if lines and not lines[-1].endswith('\n'): lines[-1] += '\n' lines.append(f'{param} = {token}\n') return start, end, values = entries[0] if token in [part for part in re.split(r'[\s,]+', ' '.join(values)) if part]: raise ConfigError(f'SRS token already present in {param}') if end > start + 1: lines.insert(end, f' {token}\n') else: line = lines[start].rstrip('\r\n') lines[start] = line.rstrip() + ' ' + token + '\n' def remove_token(lines, param, token, was_set): entries = read_entries(lines, param) if len(entries) != 1: raise ConfigError(f'Cannot safely remove token from {param}: missing parameter') first, end, _ = entries[0] indices = [] for idx in range(first, end): value = lines[idx].split('=', 1)[1] if idx == first else lines[idx] parts = [p for p in re.split(r'[\s,]+', value.strip()) if p] if token in parts: indices.append(idx) if len(indices) != 1: raise ConfigError(f'Expected exactly one SRS token in {param}') idx = indices[0] line = lines[idx] if idx == first: before, value = line.split('=', 1) remaining = [p for p in re.split(r'[\s,]+', value.strip()) if p != token] if not was_set and end == first + 1 and not remaining: del lines[first] else: lines[idx] = before.rstrip() + ' = ' + ' '.join(remaining) + '\n' else: remaining = [p for p in re.split(r'[\s,]+', line.strip()) if p != token] if remaining: lines[idx] = ' ' + ' '.join(remaining) + '\n' else: del lines[idx] if not was_set and not current_tokens(lines, param)[1]: entries = read_entries(lines, param) if entries: start, end, _ = entries[0] del lines[start:end] def write_atomic(path, content): st = os.stat(path) if not stat.S_ISREG(st.st_mode): raise ConfigError('main.cf is not a regular file') parent = os.path.dirname(os.path.abspath(path)) fd, temp_path = tempfile.mkstemp(prefix='.main.cf.srs-alias.', dir=parent) try: with os.fdopen(fd, 'w', encoding='utf-8', newline='') as f: f.writelines(content) f.flush() os.fsync(f.fileno()) os.chmod(temp_path, stat.S_IMODE(st.st_mode)) os.chown(temp_path, st.st_uid, st.st_gid) os.replace(temp_path, path) finally: if os.path.exists(temp_path): os.unlink(temp_path) def run(args): path = Path(args.main_cf) lines = path.read_text(encoding='utf-8').splitlines(keepends=True) state, extra = inspect(lines, args.domain, args.reverse_map) if args.mode == 'status': print(state) return if args.mode == 'add': if state != 'absent': raise ConfigError('Add refused: expected absent state, got ' + state) # Add in reverse order so insertion indexes remain consistent. append_token(lines, PARAMS[1], args.reverse_map) append_token(lines, PARAMS[0], args.domain) lines.append(make_marker(args.domain, args.reverse_map, extra)) elif args.mode == 'adopt': if state != 'legacy': raise ConfigError('Adoption refused: expected legacy state, got ' + state) if lines and not lines[-1].endswith('\n'): lines[-1] += '\n' lines.append(make_marker(args.domain, args.reverse_map, (True, True))) elif args.mode == 'remove': if state != 'managed': raise ConfigError('Removal refused: expected managed state, got ' + state) was_domains, was_maps = extra remove_token(lines, PARAMS[1], args.reverse_map, was_maps) remove_token(lines, PARAMS[0], args.domain, was_domains) lines[:] = [line for line in lines if not line.startswith(MARKER_PREFIX)] else: raise ConfigError('Unknown mode') expected = 'absent' if args.mode == 'remove' else 'managed' check, _ = inspect(lines, args.domain, args.reverse_map) if check != expected: raise ConfigError(f'Post-modification state unexpected: {check}') write_atomic(path, lines) print(check) def main(): parser = argparse.ArgumentParser() parser.add_argument('mode', choices=['status', 'add', 'adopt', 'remove']) parser.add_argument('--main-cf', required=True) parser.add_argument('--domain', required=True) parser.add_argument('--reverse-map', required=True) args = parser.parse_args() if not re.fullmatch(r'[A-Za-z0-9.-]+', args.domain): parser.error('Unsafe domain') if not re.fullmatch(r'socketmap:inet:127\.0\.0\.1:[0-9]{1,5}:reverse', args.reverse_map): parser.error('Unexpected reverse map') try: run(args) except (ConfigError, OSError, UnicodeError) as exc: print(f'SRS alias configuration error: {exc}', file=sys.stderr) raise SystemExit(1) if __name__ == '__main__': main()