install/mailsystem
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Extend selective SRS management and add installer tests

Browse Source
This commit is contained in:
chris
2026-10-20 11:20:38 +02:00
parent 3a7248191c
commit 53b92f7867
13 changed files with 2332 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
helpers/srs-virtual-alias.py
Executable
+223
View File
@@ -0,0 +1,223 @@
#!/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()