cpanel-importer/scripts/lib/mail-tree-links.py

#!/usr/bin/env python3
"""
mail-tree-links.py — link policy for cpmove-*/homedir/mail when the mail
tree is extracted (IMPORT_INCLUDE_MAIL=true). Called by extract.sh.

A Maildir has no legitimate symlinks. Hard links are different: Dovecot
copies a message between folders as a hard link (maildir_copy_with_hardlinks,
default yes), so a link whose name AND target are both inside the same
account's mail tree is real mail and is kept. Every other link touching a
mail tree is stripped (pre-extract) or fails the extract (post-extract).

  scan <tarball> <excludes_file> <actions_json>
      Pre-extract classifier (an optimisation; `verify` is the authority).
      Reads the archive's own member list with tarfile (exact names and
      types, no `tar -tv` text parsing) and REFUSES the archive (exit 3)
      when any member name, or hard-link target, could be read differently
      by this classifier and by tar:
        - a control character (\\x00-\\x1f, \\x7f) in any name or link
          target (a newline would split one --exclude-from pattern in two);
        - a non-canonical name: absolute, a '..' component, or
          posixpath.normpath() differing from the name after stripping one
          leading './' (e.g. 'x/./homedir', 'x/homedir//mail').
      Symlink TARGETS are not canonical-checked: cPanel archives carry
      ordinary relative links (homedir/access-logs, about -> ../public_html)
      outside the mail tree, and every symlink inside it is stripped anyway.
      Then appends a glob-escaped --exclude-from line for every symlink under
      a mail tree and every hard link that crosses a mail-tree boundary, and
      records them as 'stripped_mail_link' actions. Prints the count.

  verify <dest>
      Post-extract authority. lstat-walks every cpmove-*/homedir/mail under
      <dest> (never following links) and exits 4 when it finds a symlink, or
      a regular file whose st_nlink is larger than the number of its names
      inside that same mail tree (some other name exists outside it).
"""
import json
import os
import posixpath
import re
import stat
import sys
import tarfile

MAIL = re.compile(r'^(cpmove-[^/]+)/homedir/mail(/|$)')
CTRL = re.compile(r'[\x00-\x1f\x7f]')


def refuse(code, msg):
    sys.stderr.write(f'mail-tree-links: REFUSING archive: {msg}\n')
    sys.exit(code)


def canonical(raw):
    """The name tar will use, or None when it is not canonical."""
    n = raw[2:] if raw.startswith('./') else raw
    if n == '' or n.startswith('/') or '..' in n.split('/'):
        return None
    return n if posixpath.normpath(n) == n else None


def owner(n):
    m = MAIL.match(n)
    return m.group(1) if m else None


def scan(tar_path, excl_path, actions_path):
    try:
        with open(actions_path) as ah:
            doc = json.load(ah)
    except Exception:
        doc = {'actions': [], 'count': 0}
    already = {a.get('path') for a in doc.get('actions', [])}
    found = []
    with tarfile.open(tar_path, 'r|*') as tf:
        for m in tf:
            for label, value in (('name', m.name), ('link target', m.linkname)):
                if CTRL.search(value or ''):
                    refuse(3, f'member {label} contains a control character: {value!r}')
            name = canonical(m.name)
            if name is None:
                refuse(3, f'non-canonical member name: {m.name!r}')
            if m.issym():
                if owner(name):
                    found.append((m.name, 'symlink', m.linkname))
            elif m.islnk():
                target = canonical(m.linkname)
                if target is None:
                    refuse(3, f'non-canonical hard-link target: {m.name!r} -> {m.linkname!r}')
                a, b = owner(name), owner(target)
                if (a or b) and a != b:
                    found.append((m.name, 'hardlink', m.linkname))
    found = [f for f in found if f[0] not in already]
    with open(excl_path, 'a') as eh:
        for name, _, _ in found:
            eh.write(re.sub(r'([*?\[\\])', r'\\\1', name) + '\n')
    doc['actions'] = doc.get('actions', []) + [
        {'action': 'stripped_mail_link', 'path': name, 'target': target,
         'reason': f'{kind} inside the mail tree (Maildirs have no legitimate links outside it)'}
        for name, kind, target in found
    ]
    doc['count'] = len(doc['actions'])
    with open(actions_path, 'w') as ah:
        json.dump(doc, ah, indent=2)
    for name, kind, target in found:
        sys.stderr.write(f'  STRIP {kind}: {name} -> {target}\n')
    print(len(found))


def verify(dest):
    bad = []
    for top in sorted(os.listdir(dest)):
        mail = os.path.join(dest, top, 'homedir', 'mail')
        if not top.startswith('cpmove-'):
            continue
        try:
            st = os.lstat(mail)
        except FileNotFoundError:
            continue
        if stat.S_ISLNK(st.st_mode):
            bad.append(f'symlink: {top}/homedir/mail')
            continue
        names = {}   # (dev, ino) -> [count of names in this tree, st_nlink, a path]
        for root, dirs, files in os.walk(mail, followlinks=False):
            for e in dirs + files:
                p = os.path.join(root, e)
                s = os.lstat(p)
                if stat.S_ISLNK(s.st_mode):
                    bad.append(f'symlink: {os.path.relpath(p, dest)}')
                elif stat.S_ISREG(s.st_mode) and s.st_nlink > 1:
                    rec = names.setdefault((s.st_dev, s.st_ino), [0, s.st_nlink, p])
                    rec[0] += 1
        for count, nlink, p in names.values():
            if nlink > count:
                bad.append(f'hard link leaving the mail tree: {os.path.relpath(p, dest)} '
                           f'(st_nlink {nlink}, {count} name(s) inside the mail tree)')
    for b in bad:
        sys.stderr.write(f'mail-tree-links: {b}\n')
    if bad:
        refuse(4, f'{len(bad)} link(s) in the extracted mail tree')


if __name__ == '__main__':
    if len(sys.argv) == 5 and sys.argv[1] == 'scan':
        scan(sys.argv[2], sys.argv[3], sys.argv[4])
    elif len(sys.argv) == 3 and sys.argv[1] == 'verify':
        verify(sys.argv[2])
    else:
        sys.stderr.write('usage: mail-tree-links.py scan <tarball> <excludes> <actions_json> | verify <dest>\n')
        sys.exit(2)