Tar Archive Handler (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: Tar Archive Handler

Write a program that creates, extracts, lists, and manipulates TAR archives with support for multiple compression methods (gzip, bzip2, xz), preserving file permissions and ownership metadata.

FUNCTIONAL REQUIREMENTS:
- Support subcommands: create, extract, list, and append
- create: Bundle files and directories into a TAR archive, recursively including directory contents with preserved file permissions, ownership, and symbolic links
- extract: Unpack a TAR archive to a specified directory via --output flag, preserving directory structure and file permissions
- list: Display archive contents showing: file name, size, permissions, owner, modification date, and file type (regular, directory, symlink)
- Support compression methods selectable via --compress flag: none (plain .tar), gzip (.tar.gz), bzip2 (.tar.bz2), xz (.tar.xz); auto-detect during extraction
- append: Add files to an existing uncompressed TAR archive
- Support file filtering via --include and --exclude flags with glob patterns
- Support extracting specific files from an archive by listing their paths as additional arguments
- Verify archive integrity via --verify flag
- Print operation summary to console: total files, total size, compression ratio (if compressed), and processing time
- Save operation log as JSON with --log flag
- If no arguments are given, create a sample directory structure with files of various types, create archives in each compression format, list contents of each, extract one, and compare compression ratios
- Handle errors: corrupted archives, unsupported formats, path traversal attacks, and permission issues during extraction

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# Tar Archive Handler (Python)

## Requirements
- Ubuntu 22.04
- Python 3.10+

## Install
```bash
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
```

No external dependencies are required.

## Run
```bash
python src/main.py create archive_base input_dir --compress gzip
python src/main.py list archive_base.tar.gz
python src/main.py extract archive_base.tar.gz --output extracted
python src/main.py append archive_base.tar new_file.txt
```

If no arguments are provided, sample archives are generated and analyzed.

# No external dependencies required

#!/usr/bin/env python3
import argparse
import fnmatch
import json
import os
import shutil
import tarfile
import time
from pathlib import Path


def collect_files(p: Path):
    if p.is_file():
        return [p.resolve()]
    return [x.resolve() for x in p.rglob("*") if x.is_file()]


def matches(rel: str, includes: list[str], excludes: list[str]) -> bool:
    inc_ok = not includes or any(fnmatch.fnmatch(rel, pat) for pat in includes)
    exc = any(fnmatch.fnmatch(rel, pat) for pat in excludes)
    return inc_ok and not exc


def mode_for_create(compress: str) -> str:
    return {
        "none": "w",
        "gzip": "w:gz",
        "bzip2": "w:bz2",
        "xz": "w:xz",
    }.get(compress, "w")


def mode_for_read(path: Path) -> str:
    name = path.name.lower()
    if name.endswith(".tar.gz") or name.endswith(".tgz"):
        return "r:gz"
    if name.endswith(".tar.bz2"):
        return "r:bz2"
    if name.endswith(".tar.xz"):
        return "r:xz"
    return "r"


def default_archive_path(base: str, compress: str) -> Path:
    if compress == "gzip":
        return Path(base + ".tar.gz")
    if compress == "bzip2":
        return Path(base + ".tar.bz2")
    if compress == "xz":
        return Path(base + ".tar.xz")
    return Path(base + ".tar")


def summarize(members: list[dict]) -> dict:
    total = sum(m["size"] for m in members)
    return {"totalFiles": len(members), "totalSize": total}


def list_archive(archive: Path) -> dict:
    with tarfile.open(archive, mode_for_read(archive)) as tf:
        members = []
        for m in tf.getmembers():
            if not m.isfile() and not m.issym() and not m.isdir():
                continue
            members.append(
                {
                    "name": m.name,
                    "size": m.size,
                    "permissions": oct(m.mode),
                    "owner": f"{m.uname or m.uid}:{m.gname or m.gid}",
                    "modified": int(m.mtime),
                    "type": "symlink" if m.issym() else "directory" if m.isdir() else "regular",
                }
            )
    return {"entries": members, "summary": summarize([m for m in members if m["type"] != "directory"])}


def create_archive(args) -> dict:
    out = Path(args.output) if args.output else default_archive_path(args.archive, args.compress)
    mode = mode_for_create(args.compress)

    with tarfile.open(out, mode) as tf:
        for src in args.inputs:
            p = Path(src)
            if not p.exists():
                print(f"Warning: missing path {p}")
                continue
            base = p.parent if p.is_file() else p
            for f in collect_files(p):
                rel = f.relative_to(base).as_posix()
                if not matches(rel, args.include, args.exclude):
                    continue
                tf.add(f, arcname=rel, recursive=False)

    listed = list_archive(out)
    return {"operation": "create", "archive": str(out.resolve()), **listed}


def safe_extract(tf: tarfile.TarFile, member: tarfile.TarInfo, out_dir: Path):
    target = (out_dir / member.name).resolve()
    if not str(target).startswith(str(out_dir.resolve())):
        raise RuntimeError(f"Path traversal blocked: {member.name}")
    tf.extract(member, out_dir)


def extract_archive(args) -> dict:
    archive = Path(args.archive).resolve()
    out_dir = Path(args.output or ".").resolve()
    out_dir.mkdir(parents=True, exist_ok=True)

    selected = set(args.paths or [])
    extracted = []
    with tarfile.open(archive, mode_for_read(archive)) as tf:
        for m in tf.getmembers():
            if selected and m.name not in selected:
                continue
            safe_extract(tf, m, out_dir)
            extracted.append(m.name)

    verify = None
    if args.verify:
        with tarfile.open(archive, mode_for_read(archive)) as tf:
            verify = tf.getmembers() is not None

    return {
        "operation": "extract",
        "archive": str(archive),
        "outputDirectory": str(out_dir),
        "extractedCount": len(extracted),
        "verify": verify,
    }


def append_archive(args) -> dict:
    archive = Path(args.archive).resolve()
    if archive.suffix != ".tar":
        raise RuntimeError("append supports only uncompressed .tar archives")

    with tarfile.open(archive, "a") as tf:
        for src in args.inputs:
            p = Path(src)
            if not p.exists():
                print(f"Warning: missing path {p}")
                continue
            base = p.parent if p.is_file() else p
            for f in collect_files(p):
                rel = f.relative_to(base).as_posix()
                if not matches(rel, args.include, args.exclude):
                    continue
                tf.add(f, arcname=rel, recursive=False)

    listed = list_archive(archive)
    return {"operation": "append", "archive": str(archive), **listed}


def create_sample_data() -> Path:
    root = Path("sample_tar_data").resolve()
    if root.exists():
        shutil.rmtree(root)
    (root / "docs").mkdir(parents=True)
    (root / "assets").mkdir(parents=True)
    (root / "docs" / "readme.txt").write_text("Sample tar data\n", encoding="utf-8")
    (root / "docs" / "table.csv").write_text("id,name\n1,A\n2,B\n", encoding="utf-8")
    (root / "assets" / "blob.bin").write_bytes(os.urandom(2048))
    return root


def print_table(entries: list[dict]):
    headers = ["Name", "Size", "Perm", "Owner", "Modified", "Type"]
    rows = [[e["name"], str(e["size"]), e["permissions"], e["owner"], str(e["modified"]), e["type"]] for e in entries]
    widths = [max(len(h), *(len(r[i]) for r in rows)) for i, h in enumerate(headers)]

    def fmt(row):
        return "  ".join(v.ljust(widths[i]) for i, v in enumerate(row))

    print(fmt(headers))
    print(fmt(["-" * w for w in widths]))
    for r in rows:
        print(fmt(r))


def main() -> int:
    parser = argparse.ArgumentParser(description="Tar archive handler")
    sub = parser.add_subparsers(dest="cmd")

    c = sub.add_parser("create")
    c.add_argument("archive")
    c.add_argument("inputs", nargs="+")
    c.add_argument("--compress", choices=["none", "gzip", "bzip2", "xz"], default="none")
    c.add_argument("--include", action="append", default=[])
    c.add_argument("--exclude", action="append", default=[])
    c.add_argument("--output")
    c.add_argument("--verify", action="store_true")
    c.add_argument("--log")

    e = sub.add_parser("extract")
    e.add_argument("archive")
    e.add_argument("paths", nargs="*")
    e.add_argument("--output")
    e.add_argument("--verify", action="store_true")
    e.add_argument("--log")

    l = sub.add_parser("list")
    l.add_argument("archive")
    l.add_argument("--log")

    a = sub.add_parser("append")
    a.add_argument("archive")
    a.add_argument("inputs", nargs="+")
    a.add_argument("--include", action="append", default=[])
    a.add_argument("--exclude", action="append", default=[])
    a.add_argument("--log")

    args = parser.parse_args()

    try:
        if not args.cmd:
            root = create_sample_data()
            report = {
                "plain": create_archive(argparse.Namespace(archive="sample_plain", inputs=[str(root)], compress="none", include=[], exclude=[], output=None, verify=False, log=None)),
                "gzip": create_archive(argparse.Namespace(archive="sample_gzip", inputs=[str(root)], compress="gzip", include=[], exclude=[], output=None, verify=False, log=None)),
                "bzip2": create_archive(argparse.Namespace(archive="sample_bzip2", inputs=[str(root)], compress="bzip2", include=[], exclude=[], output=None, verify=False, log=None)),
                "xz": create_archive(argparse.Namespace(archive="sample_xz", inputs=[str(root)], compress="xz", include=[], exclude=[], output=None, verify=False, log=None)),
            }
            report["list"] = list_archive(Path(report["gzip"]["archive"]))
            report["extract"] = extract_archive(argparse.Namespace(archive=report["gzip"]["archive"], output="sample_tar_extract", paths=[], verify=True, log=None))
            Path("tar_demo_report.json").write_text(json.dumps(report, indent=2), encoding="utf-8")
            print_table(report["list"]["entries"])
            return 0

        start = time.perf_counter()
        if args.cmd == "create":
            result = create_archive(args)
        elif args.cmd == "extract":
            result = extract_archive(args)
        elif args.cmd == "list":
            result = {"operation": "list", "archive": str(Path(args.archive).resolve()), **list_archive(Path(args.archive))}
        elif args.cmd == "append":
            result = append_archive(args)
        else:
            raise RuntimeError(f"Unknown command: {args.cmd}")

        if "entries" in result:
            print_table(result["entries"])
        if "summary" in result:
            s = result["summary"]
            print(f"\nSummary: files={s['totalFiles']} totalSize={s['totalSize']}")
        print(f"Time: {(time.perf_counter() - start) * 1000:.2f} ms")

        log_path = args.log or (None if args.cmd == "list" else "tar_operation_log.json")
        if log_path:
            Path(log_path).write_text(json.dumps(result, indent=2), encoding="utf-8")
        return 0
    except Exception as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())