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), or any
      directory it cannot read+search (an uninspectable subtree fails).

  scan also cross-checks its member count against `tar -tf` and refuses on
  any difference or stream error, so nothing GNU tar extracts goes
  unclassified (verify remains the authority either way).
"""
import json
import os
import posixpath
import re
import stat
import subprocess
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 = []
    seen = 0
    try:
        tf = tarfile.open(tar_path, 'r|*')
    except Exception as e:
        refuse(3, f'archive unreadable by the classifier: {e}')
    try:
        with tf:
            for m in tf:
                seen += 1
                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))
    except Exception as e:  # TarError, EOFError, zlib.error ... (SystemExit passes through)
        refuse(3, f'archive stream error after {seen} member(s): {e}')
    # The classifier must have seen every member GNU tar will extract: a
    # stream that stops early (zero block, truncation, a header tarfile
    # skips) would leave members unclassified. Count GNU tar's own listing
    # (escape quoting = one line per member, even with a newline in a name).
    try:
        lst = subprocess.run(['tar', '--quoting-style=escape', '-tf', tar_path],
                             stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False)
    except OSError as e:
        refuse(3, f'cannot list the archive with tar: {e}')
    if lst.returncode != 0:
        refuse(3, f'tar -tf failed (rc={lst.returncode}): {lst.stderr.decode(errors="replace").strip()[:300]}')
    tar_count = lst.stdout.count(b'\n')
    if tar_count != seen:
        refuse(3, f'member count differs: classifier saw {seen}, tar lists {tar_count}')
    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 = []

    def walk_error(err):
        bad.append(f'cannot inspect: {os.path.relpath(getattr(err, "filename", "") or dest, dest)} ({err.strerror})')

    try:
        tops = sorted(os.listdir(dest))
    except OSError as e:
        refuse(4, f'cannot list the extract: {e}')
    for top in tops:
        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
        if stat.S_ISDIR(st.st_mode) and not os.access(mail, os.R_OK | os.X_OK):
            bad.append(f'directory not readable+searchable: {top}/homedir/mail')
            continue
        names = {}   # (dev, ino) -> [count of names in this tree, st_nlink, a path]
        # Fail closed: os.walk silently skips what it can't read unless told.
        for root, dirs, files in os.walk(mail, followlinks=False, onerror=walk_error):
            for e in dirs + files:
                p = os.path.join(root, e)
                try:
                    s = os.lstat(p)
                except OSError as err:
                    walk_error(err)
                    continue
                if stat.S_ISLNK(s.st_mode):
                    bad.append(f'symlink: {os.path.relpath(p, dest)}')
                elif stat.S_ISDIR(s.st_mode) and not os.access(p, os.R_OK | os.X_OK):
                    bad.append(f'directory not readable+searchable: {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)} problem(s) in the extracted mail tree (links or uninspectable directories)')


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)