wpressarc

#!/usr/bin/env python3

# MIT License

# Copyright (c) 2022 André Kugland
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

"""
wpressarc - Convert ai1wm wpress archives to and from tar archives.

This script converts between the wpress archive format and the tar
archive format. It is intended to be used as a filter, so it reads from
standard input and writes to standard output.

The wpress archive format is a simple format for storing files in a
directory structure, used by the All-in-One WP Migration plugin.

Usage: wpressarc --from-tar | --to-tar [OPTIONS] [FILE ...] < INPUT

One of the following options must be specified:

  -f | --from-tar       Convert from tar to wpress archive
  -t | --to-tar         Convert from wpress to tar archive

When converting from wpress to tar, the following options are also
available:

  -m | --mode MODE      File mode for tar archive entries [default: 0644]
  -d | --dmode MODE     Directory mode for tar archive entries [default: 0755]
  -u | --uid UID        User ID for tar archive entries [default: 0]
  -g | --gid GID        Group ID for tar archive entries [default: 0]
  -U | --owner OWNER    User name for tar archive entries [default: root]
  -G | --group GROUP    Group name for tar archive entries [default: root]

For example, to convert a tar archive to a wpress archive:

    wpressarc --from-tar < wordpress.tar > wordpress.wpress

To convert a wpress archive to a tar archive:

    wpressarc --to-tar < wordpress.wpress > wordpress.tar

To extract a wpress archive:

    wpressarc --to-tar < wordpress.wpress | tar -xvf -

To extract only files matching a pattern from an archive (wpress or tar):

    wpressarc --to-tar '*.sql' < wordpress.wpress | tar -xvf -

To list the contents of a wpress archive:

    wpressarc --to-tar < wordpress.wpress | tar -tvf -

Converting a wpress archive to a tar archive is lossy, because the
wpress archive format does not store file permission, ownership, and
other metadata, and it also does not store any metadata for the
directories (in fact, it does not even store the directory entries).

When the FILE arguments are specified, they are interpreted as patterns
to match against the file names in the archive. If stdin is seekable,
i.e. if it is redirected directly from a file, then the unmatched
entries will be skipped entirely, which will be much faster than reading
and discarding the entries.

This script is licensed under the MIT license. See the LICENSE file for
details.
"""

from __future__ import annotations
import sys
import os
import tarfile
from fnmatch import fnmatch
from typing import IO


class EntryHeader:
    """Header for a single entry in a WordPress archive."""

    path: str
    name: str
    size: int
    mtime: int

    def __init__(self, path: str, name: str, size: int, mtime: int):
        """Create a new entry header."""
        self.path = path
        self.name = name
        self.size = size
        self.mtime = mtime

    def _write_field(self, file: IO[bytes], field_size: int, value: str | int):
        """Write a field to the archive."""
        value_bytes = str(value).encode("utf-8")
        if len(value_bytes) > field_size:
            raise ValueError("Field value too long")
        file.write(value_bytes)
        file.write(b"\0" * (field_size - len(value_bytes)))

    def write_header(self, file: IO[bytes]):
        """Write the header to the archive."""
        self._write_field(file, 255, self.name)
        self._write_field(file, 14, self.size)
        self._write_field(file, 12, self.mtime)
        self._write_field(file, 4096, self.path)

    def to_tarinfo(self, args: argparse.Namespace) -> tarfile.TarInfo:
        """Convert the header to a TarInfo object."""
        tarinfo = tarfile.TarInfo()
        name = os.path.join(self.path, self.name)
        if name.startswith("./"):
            name = name[2:]
        if name.startswith("/"):
            name = name[1:]
        tarinfo.name = name
        tarinfo.size = self.size
        tarinfo.mtime = self.mtime
        tarinfo.type = tarfile.REGTYPE
        tarinfo.mode = args.mode
        tarinfo.uid = args.uid
        tarinfo.gid = args.gid
        tarinfo.uname = args.owner
        tarinfo.gname = args.group
        return tarinfo

    @staticmethod
    def read_header(file: IO[bytes]) -> EntryHeader | None:
        """Read a header from the archive."""
        name = file.read(255).decode("utf-8").rstrip("\0")
        size = file.read(14).decode("ascii").rstrip("\0")
        mtime = file.read(12).decode("ascii").rstrip("\0")
        path = file.read(4096).decode("utf-8").rstrip("\0")
        if name == "" and size == "" and mtime == "" and path == "":
            return None
        else:
            return EntryHeader(path, name, int(size), int(mtime))

    @staticmethod
    def from_tarinfo(tarinfo: tarfile.TarInfo) -> EntryHeader | None:
        """Convert a TarInfo object to a header."""
        if tarinfo.type != tarfile.REGTYPE:  # Skip directories, etc.
            return None
        path = os.path.dirname(tarinfo.name)
        if path.startswith("./"):
            path = path[2:]
        path = "." if path == "" else path
        name = os.path.basename(tarinfo.name)
        return EntryHeader(path, name, tarinfo.size, tarinfo.mtime)


class Archive:
    """Reader for a WordPress archive."""

    file: IO[bytes]

    def __init__(self, file: IO[bytes]):
        """Create a new archive reader."""
        self.file = file

    def next_entry(self) -> EntryHeader | None:
        """Read the next entry header from the archive."""
        return EntryHeader.read_header(self.file)

    @staticmethod
    def _copy(input: IO[bytes], output: IO[bytes] | None, size: int):
        """Copy the contents of a file to another file."""
        if output is None and input.seekable():
            input.seek(size, os.SEEK_CUR)
        else:
            while size > 0:
                chunk = input.read(min(size, 65536))
                if len(chunk) == 0:
                    raise EOFError("Unexpected end of file")
                if output is not None:
                    output.write(chunk)
                size -= len(chunk)

    def write(self, entry: EntryHeader, input: IO[bytes]):
        """Write a file to the archive."""
        entry.write_header(self.file)
        self._copy(input, self.file, entry.size)

    def skip(self, entry: EntryHeader):
        """Skip a file in the archive."""
        self._copy(self.file, None, entry.size)

    def finalize(self):
        """Finalize the archive."""
        self.file.write(b"\0" * (255 + 14 + 12 + 4096))
        self.file.flush()


def match_fnames(fnames: list[str], path: str, name: str) -> bool:
    """Check if a file name matches any of the patterns."""
    if not fnames:  # No patterns, match everything
        return True
    if any(fnmatch(os.path.join(path, name), pattern) for pattern in fnames):
        return True
    return False


def to_tar(stdin: IO[bytes], stdout: IO[bytes], args: argparse.Namespace):
    """Convert a WordPress archive to a tar archive."""
    with tarfile.open(fileobj=stdout, mode=("w" if stdout.seekable() else "w|")) as tar:
        archive = Archive(stdin)
        added_directories: set[str] = set()
        while True:
            entry = archive.next_entry()
            if entry is None:
                break
            if not match_fnames(args.fnames, entry.path, entry.name):
                archive.skip(entry)
                continue
            tarinfo = entry.to_tarinfo(args)
            parent = os.path.dirname(tarinfo.name) + "/"
            if parent not in ["/", "./"] and parent not in added_directories:
                parent_tarinfo = tarfile.TarInfo()
                parent_tarinfo.name = parent
                parent_tarinfo.type = tarfile.DIRTYPE
                parent_tarinfo.mode = args.dmode
                parent_tarinfo.uid = args.uid
                parent_tarinfo.gid = args.gid
                parent_tarinfo.uname = args.owner
                parent_tarinfo.gname = args.group
                tar.addfile(parent_tarinfo)
                added_directories.add(parent)
            tar.addfile(tarinfo, fileobj=archive.file)
        tar.close()


def from_tar(stdin: IO[bytes], stdout: IO[bytes]):
    """Convert a tar archive to a WordPress archive."""
    with tarfile.open(fileobj=stdin, mode=("r" if stdin.seekable() else "r|")) as tar:
        archive = Archive(stdout)
        for tarinfo in tar:
            header = EntryHeader.from_tarinfo(tarinfo)
            if header is not None:  # Skip directories, etc.
                if not match_fnames(args.fnames, header.path, header.name):
                    continue
                file = tar.extractfile(tarinfo)
                if file is not None:
                    archive.write(header, file)
        archive.finalize()


def show_help(error: bool = False):
    """Show the help text."""
    print(__doc__.strip(), file=sys.stderr if error else sys.stdout)
    sys.exit(1 if error else 0)


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(prog="wpressarc", add_help=False)
    parser.add_argument("--to-tar", "-t", action="store_true")
    parser.add_argument("--from-tar", "-f", action="store_true")
    parser.add_argument("--uid", "-u", type=int)
    parser.add_argument("--gid", "-g", type=int)
    parser.add_argument("--owner", "-U")
    parser.add_argument("--group", "-G")
    parser.add_argument("--mode", "-m")
    parser.add_argument("--dmode", "-d")
    parser.add_argument("--help", "-h", action="store_true")
    parser.add_argument("fnames", nargs="*")

    # Parse the command line arguments
    args = parser.parse_args()

    if args.help:
        show_help(error=False)

    if [args.to_tar, args.from_tar].count(True) != 1:
        show_help(error=True)

    stdin = os.fdopen(sys.stdin.fileno(), "rb")
    stdout = os.fdopen(sys.stdout.fileno(), "wb")

    try:
        if args.from_tar:
            if [args.uid, args.gid, args.owner, args.group, args.mode].count(None) != 5:
                parser.error(
                    "the following arguments are not allowed in combination with --from-tar:\n"
                    "--uid, --gid, --owner, --group, --mode, --dmode"
                )
            from_tar(stdin, stdout)
        elif args.to_tar:
            args.uid = args.uid or 0
            args.gid = args.gid or 0
            args.owner = args.owner or "root"
            args.group = args.group or "root"
            (args.mode, args.dmode) = (args.mode or "644", args.dmode or "755")
            args.mode = int(args.mode, 8) & 0o777  # No set[ug]id or sticky bits :-)
            args.dmode = int(args.dmode, 8) & 0o777
            to_tar(stdin, stdout, args)
        else:
            show_help(error=True)
    finally:
        stdin.close()
        stdout.close()