224 lines
8.5 KiB
Python
Executable File
224 lines
8.5 KiB
Python
Executable File
#!/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()
|