feat: add archive plugin for zip, tar and gzip comments

README.md

@@ -109,6 +109,7 @@ $ otag write --clear id3 song.mp3
 | `pdf` | `.pdf` | The keywords field |
 | `office` | `.docx` `.xlsx` `.pptx` `.odt` `.ods` `.odp`, reading also `.doc` `.xls` `.ppt` | The description field |
 | `ebook` | `.epub` `.fb2` `.cbz` | One otag line where other apps don't look: a `<meta name="otag">` in the EPUB package file, a `<custom-info info-type="otag">` in FB2, the archive comment of CBZ (never overwriting another tool's comment); the book's own metadata is left untouched |
+| `archive` | `.zip` `.tar` `.gz` `.tgz` (so also `.tar.gz`) | An `otag:` comment that unpacking ignores: the archive comment of ZIP, the comment field of the gzip header, a pax global header at the start of a tar. Another tool's comment, like the commit id stored by `git archive`, is never replaced. Writing a tar or gzip file copies it once, the packed data stays unchanged |
 | `frontmatter` | `.md` `.markdown` `.txt` | `projects`, `contexts`, `tags` and `note` keys of the YAML front matter; other keys, comments and formatting are left untouched |
 | `comment` | Source code such as `.py` `.sh` `.toml` `.js` `.c` `.go` `.rs` `.php` `.html` | An `otag: +project @context #tag note` comment line within the first ten lines, written in the file's comment style below any shebang, encoding, `<!DOCTYPE>` or `<?php` line |
 

orvanitag/cmd.py

@@ -240,7 +240,7 @@ def parser() -> ArgumentParser:
     no_recursive(sub)
 
     sub = command('write', ['w'], 'write the entry of a file into the file itself', run_write)
-    sub.add_argument('plugin', help='comment, ebook, frontmatter, id3, office or pdf')
+    sub.add_argument('plugin', help='archive, comment, ebook, frontmatter, id3, office or pdf')
     file_arg(sub)
     sub.add_argument('-c', '--clear', action='store_true', help="remove the plugin's tags from the file instead")
 

orvanitag/orvani.py

@@ -18,6 +18,7 @@ SEPARATOR    = ' --- ' # between file name and contexts, in context files and ex
 DEFAULT_PLUGINS = [
     'frontmatter',
     'comment',
+    'archive',
     'ebook',
     'id3',
     'office',

orvanitag/plugins/_zip.py

@@ -1,5 +1,5 @@
 """
-Shared by the plugins for zip based formats; not a plugin itself.
+Shared by the plugins for zip based formats and archive comments; not a plugin itself.
 """
 import os
 import shutil
@@ -7,6 +7,8 @@ from pathlib import Path
 from zipfile import ZipFile
 from orvanitag.errors import UnsupportedFile
 
+MARKER = 'otag:' # starts the archive comments otag writes
+
 
 def rewrite(path, changes: dict = None, comment: bytes = None):
     """
@@ -35,3 +37,35 @@ def rewrite(path, changes: dict = None, comment: bytes = None):
         os.replace(tmp, path)
     finally:
         tmp.unlink(missing_ok=True)
+
+
+def unmark(comment: str) -> str:
+    """The otag line of an archive comment, or '' if the comment is not otag's"""
+    return comment.removeprefix(MARKER) if comment.startswith(MARKER) else ''
+
+
+def mark(comment: str, line: str, path) -> str | None:
+    """
+    The new archive comment for an otag line, or None to leave the comment as it is.
+    An empty line clears otag's comment; another tool's comment is never replaced.
+    """
+    ours = comment.startswith(MARKER)
+    if not line:
+        return '' if ours else None
+    if comment and not ours:
+        raise UnsupportedFile(f'the archive comment of {path} is used by another tool')
+    return f'{MARKER} {line}'
+
+
+def read_comment(path) -> str:
+    """The otag line of a zip file's archive comment"""
+    with ZipFile(path) as archive:
+        return unmark(archive.comment.decode('utf-8', 'replace'))
+
+
+def write_comment(path, line: str):
+    """Stores an otag line as the zip file's archive comment; an empty line clears it"""
+    with ZipFile(path) as archive:
+        comment = mark(archive.comment.decode('utf-8', 'replace'), line, path)
+    if comment is not None:
+        rewrite(path, comment=comment.encode('utf-8'))

orvanitag/plugins/archive.py

@@ -0,0 +1,122 @@
+"""
+Tags in archives as an `otag:` comment, leaving the packed files untouched: the archive comment of ZIP,
+the comment field of the gzip header (so also of `.tar.gz`) and a pax global header at the start of a tar.
+Another tool's comment, like the commit id `git archive` stores, is never replaced.
+"""
+import os
+import shutil
+import tarfile
+from orvanitag.plugins._zip import mark, unmark, read_comment, write_comment
+from orvanitag.entry import Entry
+from orvanitag.errors import UnsupportedFile
+
+NAME = 'archive'
+TYPES = ['.zip', '.tar', '.gz', '.tgz']
+
+# gzip header flags, RFC 1952
+FHCRC, FEXTRA, FNAME, FCOMMENT = 2, 4, 8, 16
+BLOCK = 512 # tar block size
+
+
+def until_zero(file) -> bytes:
+    """Reads a zero terminated gzip header field, including the zero"""
+    data = bytearray()
+    while not data.endswith(b'\0'):
+        byte = file.read(1)
+        if not byte:
+            raise UnsupportedFile(f'{file.name} has a truncated gzip header')
+        data += byte
+    return bytes(data)
+
+
+def gzip_header(path) -> tuple[bytes, str, int]:
+    """The gzip header without comment, the comment and where the compressed data starts"""
+    with open(path, 'rb') as file:
+        head = bytearray(file.read(10))
+        if head[:3] != b'\x1f\x8b\x08' or len(head) < 10:
+            raise UnsupportedFile(f'{path} is no gzip file')
+        flags = head[3]
+        # The header checksum is optional, so it is dropped instead of recomputed
+        head[3] = flags & ~(FCOMMENT | FHCRC)
+        if flags & FEXTRA:
+            size = file.read(2)
+            head += size + file.read(int.from_bytes(size, 'little'))
+        if flags & FNAME:
+            head += until_zero(file)
+        # RFC 1952 asks for Latin-1, but tags may use any script
+        comment = until_zero(file)[:-1].decode('utf-8', 'replace') if flags & FCOMMENT else ''
+        if flags & FHCRC:
+            file.read(2)
+        return bytes(head), comment, file.tell()
+
+
+def pax_globals(path) -> tuple[dict, int]:
+    """The records of the pax global header a tar starts with, and where the rest of the archive begins"""
+    try:
+        tarfile.open(path, 'r:').close()
+    except tarfile.TarError as error:
+        raise UnsupportedFile(f'{path} is no uncompressed tar: {error}') from error
+    with open(path, 'rb') as file:
+        header = file.read(BLOCK)
+        if len(header) < BLOCK or header[156:157] != tarfile.XGLTYPE:
+            return {}, 0
+        size = int(header[124:136].strip(b'\0 ') or b'0', 8)
+        data = file.read(size)
+    records = {}
+    while data: # records are `<length> <key>=<value>\n`, the length counting the whole record
+        length = data.split(b' ', 1)[0]
+        key, _, value = data[len(length) + 1:int(length) - 1].partition(b'=')
+        records[key.decode('utf-8')] = value.decode('utf-8')
+        data = data[int(length):]
+    return records, BLOCK + -(-size // BLOCK) * BLOCK
+
+
+def replace_head(path, end: int, head: bytes):
+    """Atomically replaces the first `end` bytes of a file by head, copying the rest as it is"""
+    tmp = path.with_name(f'.{path.name}.tmp')
+    try:
+        with open(path, 'rb') as source, open(tmp, 'wb') as target:
+            source.seek(end)
+            target.write(head)
+            shutil.copyfileobj(source, target)
+        shutil.copymode(path, tmp)
+        os.replace(tmp, path)
+    finally:
+        tmp.unlink(missing_ok=True)
+
+
+def read(path, config):
+    suffix = path.suffix.lower()
+    if suffix == '.zip':
+        text = read_comment(path)
+    elif suffix == '.tar':
+        text = unmark(pax_globals(path)[0].get('comment', ''))
+    else:
+        text = unmark(gzip_header(path)[1])
+    if text.strip():
+        return Entry.parse(text.split())
+
+
+def write(path, entry, config):
+    suffix = path.suffix.lower()
+    line = str(entry)
+    if suffix == '.zip':
+        write_comment(path, line)
+
+    elif suffix == '.tar':
+        records, end = pax_globals(path)
+        comment = mark(records.pop('comment', ''), line, path)
+        if comment is None:
+            return
+        if comment:
+            records['comment'] = comment
+        replace_head(path, end, tarfile.TarInfo.create_pax_global_header(records) if records else b'')
+
+    else:
+        head, comment, end = gzip_header(path)
+        comment = mark(comment, line, path)
+        if comment is None:
+            return
+        if comment:
+            head = head[:3] + bytes([head[3] | FCOMMENT]) + head[4:] + comment.encode('utf-8') + b'\0'
+        replace_head(path, end, head)

orvanitag/plugins/ebook.py

@@ -8,13 +8,12 @@ import re
 import shutil
 from xml.sax.saxutils import escape, unescape
 from zipfile import ZipFile
-from orvanitag.plugins._zip import rewrite
+from orvanitag.plugins._zip import rewrite, read_comment, write_comment
 from orvanitag.entry import Entry
 from orvanitag.errors import UnsupportedFile
 
 NAME = 'ebook'
 TYPES = ['.epub', '.fb2', '.cbz']
-MARKER = 'otag:' # starts the CBZ archive comment
 QUOTES = {'"': '"'}
 
 # Element prefixes like `opf:` are matched, so the new element can use the same namespace prefix
@@ -59,9 +58,7 @@ def read(path, config):
         found = CUSTOM.search(read_fb2(path)[0])
         text = unescape(found.group(1)) if found else ''
     elif suffix == '.cbz':
-        with ZipFile(path) as archive:
-            comment = archive.comment.decode('utf-8', 'replace')
-        text = comment.removeprefix(MARKER) if comment.startswith(MARKER) else ''
+        text = read_comment(path)
     if text.strip():
         return Entry.parse(text.split())
 
@@ -114,13 +111,4 @@ def write(path, entry, config):
             tmp.unlink(missing_ok=True)
 
     elif suffix == '.cbz':
-        with ZipFile(path) as archive:
-            comment = archive.comment
-        ours = comment.startswith(MARKER.encode())
-        if not entry:
-            if ours: # another tool's comment is not otag's to clear
-                rewrite(path, comment=b'')
-        elif comment and not ours:
-            raise UnsupportedFile(f'the archive comment of {path} is used by another tool')
-        else:
-            rewrite(path, comment=f'{MARKER} {line}'.encode('utf-8'))
+        write_comment(path, line)

tests/test_archive.py

@@ -0,0 +1,139 @@
+import gzip
+import io
+import struct
+import tarfile
+import zipfile
+import zlib
+
+import pytest
+
+import orvanitag.orvani as orvani
+from orvanitag.entry import Entry
+from orvanitag.orvani import OrvaniTag
+from orvanitag.errors import UnsupportedFile
+
+
+@pytest.fixture
+def cwd(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+    return tmp_path
+
+
+def make_tar(path, **kwargs):
+    with tarfile.open(path, 'w', **kwargs) as archive:
+        data = b'packed'
+        info = tarfile.TarInfo('a.txt')
+        info.size = len(data)
+        archive.addfile(info, io.BytesIO(data))
+
+
+def members(path):
+    with tarfile.open(path) as archive:
+        return {member.name: archive.extractfile(member).read() for member in archive.getmembers()}
+
+
+def roundtrip(cwd, name, *tags):
+    """Writes the tags, reads them back into a fresh entry and clears them again"""
+    OrvaniTag().add(name, *tags)
+    OrvaniTag().write_metadata(name, 'archive')
+    expected = (cwd / orvani.CONTEXT_FILE).read_text()
+    (cwd / orvani.CONTEXT_FILE).unlink()
+    OrvaniTag().read_metadata(name)
+    assert (cwd / orvani.CONTEXT_FILE).read_text() == expected
+
+
+def test_zip_comment(cwd):
+    with zipfile.ZipFile(cwd / 'a.zip', 'w') as z:
+        z.writestr('a.txt', 'packed')
+    roundtrip(cwd, 'a.zip', '+backup', '#old')
+    with zipfile.ZipFile(cwd / 'a.zip') as z:
+        assert z.comment == b'otag: +backup #old'
+        assert z.read('a.txt') == b'packed'
+
+
+@pytest.mark.parametrize('name', ['a.gz', 'a.tar.gz', 'a.tgz'])
+def test_gzip_comment_and_clear_restores_the_file(cwd, name):
+    with gzip.GzipFile(cwd / name, 'wb', mtime=0) as file:
+        file.write(b'packed')
+    before = (cwd / name).read_bytes()
+
+    roundtrip(cwd, name, '#фантастика', 'a note')
+    assert b'otag: #\xd1\x84' in (cwd / name).read_bytes()
+    assert gzip.decompress((cwd / name).read_bytes()) == b'packed'
+
+    OrvaniTag().write_metadata(name, 'archive', clear=True)
+    assert (cwd / name).read_bytes() == before
+
+
+def test_gzip_keeps_extra_field_and_drops_header_crc(cwd):
+    header = b'\x1f\x8b\x08' + bytes([2 | 4 | 8]) + bytes(6) + struct.pack('<H', 4) + b'ab\0\0' + b'a.txt\0'
+    deflate = zlib.compressobj(wbits=-15)
+    body = deflate.compress(b'packed') + deflate.flush() + struct.pack('<II', zlib.crc32(b'packed'), 6)
+    (cwd / 'a.gz').write_bytes(header + struct.pack('<H', zlib.crc32(header) & 0xffff) + body)
+
+    roundtrip(cwd, 'a.gz', '#lofi')
+    data = (cwd / 'a.gz').read_bytes()
+    assert data == b'\x1f\x8b\x08' + bytes([4 | 8 | 16]) + header[4:] + b'otag: #lofi\0' + body
+    assert gzip.decompress(data) == b'packed'
+
+
+def test_tar_global_header_and_clear_restores_the_file(cwd):
+    make_tar(cwd / 'a.tar')
+    before = (cwd / 'a.tar').read_bytes()
+
+    roundtrip(cwd, 'a.tar', '+backup', 'a note')
+    with tarfile.open(cwd / 'a.tar') as archive:
+        assert archive.pax_headers == {'comment': 'otag: +backup a note'}
+    assert members(cwd / 'a.tar') == {'a.txt': b'packed'}
+
+    OrvaniTag().add('a.tar', '#more')
+    OrvaniTag().write_metadata('a.tar', 'archive')
+    with tarfile.open(cwd / 'a.tar') as archive:
+        assert archive.pax_headers == {'comment': 'otag: +backup #more a note'}
+
+    OrvaniTag().write_metadata('a.tar', 'archive', clear=True)
+    assert (cwd / 'a.tar').read_bytes() == before
+
+
+def test_tar_keeps_other_global_records(cwd):
+    make_tar(cwd / 'a.tar', pax_headers={'mtime': '1'})
+    OrvaniTag().add('a.tar', '#lofi')
+    OrvaniTag().write_metadata('a.tar', 'archive')
+    OrvaniTag().write_metadata('a.tar', 'archive', clear=True)
+    with tarfile.open(cwd / 'a.tar') as archive:
+        assert archive.pax_headers == {'mtime': '1'}
+    assert members(cwd / 'a.tar') == {'a.txt': b'packed'}
+
+
+def foreign_zip(path):
+    with zipfile.ZipFile(path, 'w') as z:
+        z.comment = b'foreign'
+
+
+def foreign_gzip(path):
+    path.write_bytes(b'\x1f\x8b\x08' + bytes([16]) + bytes(6) + b'foreign\0' + gzip.compress(b'packed', mtime=0)[10:])
+
+
+@pytest.mark.parametrize('name, make', [
+    ('a.tar', lambda path: make_tar(path, pax_headers={'comment': '0123abcd'})), # like git archive
+    ('a.gz', foreign_gzip),
+    ('a.zip', foreign_zip),
+])
+def test_foreign_comments_are_kept(cwd, name, make):
+    make(cwd / name)
+    before = (cwd / name).read_bytes()
+    assert OrvaniTag().read_metadata(name) == Entry()
+    OrvaniTag().add(name, '#lofi')
+    with pytest.raises(UnsupportedFile, match='used by another tool'):
+        OrvaniTag().write_metadata(name, 'archive')
+    OrvaniTag().write_metadata(name, 'archive', clear=True)
+    assert (cwd / name).read_bytes() == before
+
+
+@pytest.mark.parametrize('name', ['a.tar', 'a.gz'])
+def test_other_files_are_refused_unchanged(cwd, name):
+    (cwd / name).write_bytes(gzip.compress(b'packed') if name == 'a.tar' else b'plain text')
+    OrvaniTag().add(name, '#lofi')
+    with pytest.raises(UnsupportedFile):
+        OrvaniTag().write_metadata(name, 'archive')
+    assert OrvaniTag().scan('.', False)[1]