fix(deploy): parse backup connection aliases and quoted values safely

2026-11-02 12:11:48 +01:00
parent 653f8d04d4
commit f19e7ebf4a
5 changed files with 191 additions and 11 deletions
@@ -0,0 +1,68 @@
+"""Read a deployment connection string from stdin; emit five fields plus a sentinel.
+
+This is a single-host dump contract, not a replacement for MySqlConnector. Reject
+ambiguous aliases/control characters instead of choosing a different backup target.
+Never include input values in errors or pass the connection string as an argument.
+"""
+import re
+import sys
+
+
+ALIASES = {
+    **dict.fromkeys(['server', 'host', 'data source', 'datasource', 'address', 'addr', 'network address'], 'host'),
+    **dict.fromkeys(['user id', 'userid', 'username', 'uid', 'user name', 'user'], 'user'),
+    **dict.fromkeys(['password', 'pwd'], 'password'),
+    **dict.fromkeys(['database', 'initial catalog'], 'database'),
+    'port': 'port',
+}
+PAIR = re.compile(r'''\s*([^=;]+?)\s*=\s*(?:"((?:""|[^"])*)"|'((?:''|[^'])*)'|([^;]*))\s*(?:;|$)''')
+
+
+def parse(connection):
+    if any(ord(c) < 32 or ord(c) == 127 for c in connection):
+        raise ValueError('Control characters are unsupported in backup connection settings.')
+    fields = {}
+    position = 0
+    while position < len(connection):
+        if connection[position] == ';' or connection[position].isspace():
+            position += 1
+            continue
+        match = PAIR.match(connection, position)
+        if not match:
+            raise ValueError('Malformed backup connection settings.')
+        key, double, single, plain = match.groups()
+        if double is not None:
+            value = double.replace('""', '"')
+        elif single is not None:
+            value = single.replace("''", "'")
+        else:
+            value = plain.strip()
+            if value.startswith(('"', "'")):
+                raise ValueError('Malformed quoted backup connection value.')
+        canonical = ALIASES.get(key.strip().lower())
+        if canonical:
+            if canonical in fields:
+                raise ValueError('Duplicate backup connection setting or alias.')
+            fields[canonical] = value
+        position = match.end()
+    if any(not fields.get(key, '').strip() for key in ['host', 'database', 'user']):
+        raise ValueError('Backup connection requires host, database and user.')
+    port = fields.get('port', '3306')
+    if not re.fullmatch(r'[0-9]+', port) or not 1 <= int(port) <= 65535:
+        raise ValueError('Backup connection port must be between 1 and 65535.')
+    if any(c in fields['host'] for c in ',/\\'):
+        raise ValueError('Backup requires one TCP host, not a host list or socket path.')
+    if any(c in fields['database'] for c in '/\\'):
+        raise ValueError('Database name cannot contain backup path separators.')
+    return [fields['host'], str(int(port)), fields['database'], fields['user'], fields.get('password', '')]
+
+
+if __name__ == '__main__':
+    try:
+        values = parse(sys.stdin.buffer.read().decode('utf-8'))
+    except (ValueError, UnicodeError):
+        print('Invalid or unsupported backup connection settings.', file=sys.stderr)
+        sys.exit(1)
+    sys.stdout.reconfigure(encoding='utf-8', newline='\n')
+    # Sentinel keeps an empty final password intact through shell command substitution.
+    sys.stdout.write('\n'.join([*values, 'END']) + '\n')