GVSDS/Viper_VRTE
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
Viper_VRTE/tarfile.vp
T

1010 lines
37 KiB
Plaintext
Raw Permalink Blame History

# tarfile - tar archive reader and writer (POSIX ustar, PAX and GNU extensions).
#
# Read: ustar headers with the prefix field, GNU 'L'/'K' long name/link headers, PAX 'x'/'g'
# extended headers (path, linkpath, size, mtime, uid, gid, uname, gname) and GNU base-256 numeric
# fields. The archive is read into memory as a whole, so a .tar.gz is read by inflating first.
# Write: ustar headers, plus a PAX 'x' record when a path or link target does not fit ustar fields.
#
# Usage:
# r: tarfile.TarFile* = tarfile.TarFile(path, "r")
# if r.ok == 0: ... # tarfile.error() carries the reason
# m: tarfile.TarInfo* = r.first()
# while m is not None: ... # m.name / m.size / m.typeflag
# m = m.next
# r.extractall("out")
# r.close()
#
# w: tarfile.TarFile* = tarfile.TarFile(path, "w")
# w.add_dir("a")
# w.add_file("a/b.txt", data, data_len, 420, 0)
# w.close() # writes the end-of-archive blocks
import stdlib
import string
import win32.win32base as wb
import win32.win32file as wf
from stdint import *
# ============================================================
# constants
# ============================================================
define TAR_BLOCK = 512
define TAR_NAME_MAX = 100
define TAR_PREFIX_MAX = 155
define TAR_LINKNAME_MAX = 100
# header field offsets
define TAR_OFF_NAME = 0
define TAR_OFF_MODE = 100
define TAR_OFF_UID = 108
define TAR_OFF_GID = 116
define TAR_OFF_SIZE = 124
define TAR_OFF_MTIME = 136
define TAR_OFF_CHKSUM = 148
define TAR_OFF_TYPEFLAG = 156
define TAR_OFF_LINKNAME = 157
define TAR_OFF_MAGIC = 257
define TAR_OFF_UNAME = 265
define TAR_OFF_GNAME = 297
define TAR_OFF_DEVMAJOR = 329
define TAR_OFF_DEVMINOR = 337
define TAR_OFF_PREFIX = 345
# typeflag values
define TAR_REG = '0'
define TAR_REG_OLD = 0
define TAR_HARDLINK = '1'
define TAR_SYMLINK = '2'
define TAR_DIR = '5'
define TAR_CONT = '7'
define TAR_PAX_NEXT = 'x'
define TAR_PAX_GLOBAL = 'g'
define TAR_GNU_LONGNAME = 'L'
define TAR_GNU_LONGLINK = 'K'
# write format
define TAR_FORMAT_USTAR = 0
define TAR_FORMAT_PAX = 1
# permission bits
define TAR_MODE_FILE = 420 # 0644
define TAR_MODE_DIR = 493 # 0755
define TAR_MODE_LINK = 511 # 0777
# PAX key bits
define PAX_PATH = 1
define PAX_LINKPATH = 2
define PAX_UNAME = 4
define PAX_GNAME = 8
define PAX_SIZE = 16
define PAX_MTIME = 32
define PAX_UID = 64
define PAX_GID = 128
# error codes
define TAR_OK = 0
define TAR_ERR_OPEN = -1
define TAR_ERR_FORMAT = -2
define TAR_ERR_MEMORY = -3
define TAR_ERR_IO = -4
define TAR_ERR_CHECKSUM = -5
define TAR_ERR_UNSUPPORTED = -6
define TAR_ERR_MODE = -7
define TAR_ERR_CLOSED = -8
define TAR_ERR_UNSAFE = -9
# handles are opened shared so a concurrent reader does not lock the archive out
define TAR_SHARE = 7
# ============================================================
# module error state
# ============================================================
tarfile_error_code_val: int = 0
tarfile_error_msg: [256]char = ""
def tar_set_error(code: int, msg: str):
global tarfile_error_code_val, tarfile_error_msg
tarfile_error_code_val = code
string.strncpy(tarfile_error_msg, msg, 255)
tarfile_error_msg[255] = 0
def error_code() -> int:
return tarfile_error_code_val
def error() -> str:
return tarfile_error_msg
# ============================================================
# numeric header fields
# ============================================================
# Parse a numeric field: leading spaces and NULs are skipped, then octal digits are consumed up to
# the first non-octal byte. A first byte with bit 7 set selects the GNU base-256 form.
def tar_num(p: uint8*, n: sizet) -> uint64:
if n == 0: return 0
if (p[0] & 128) != 0:
v: int64 = int64(p[0] & 127)
for i in range(1, int(n)):
v = (v << 8) | int64(p[i])
if v < 0: return 0
return uint64(v)
out: uint64 = 0
i: sizet = 0
while i < n and (p[i] == 32 or p[i] == 0): i += 1
while i < n:
c: uint8 = p[i]
if c < '0' or c > '7': break
out = (out << 3) + uint64(c - '0')
i += 1
return out
# Write an n-byte numeric field as octal with a trailing NUL. A value past the octal range falls
# back to the GNU base-256 form: first byte 0x80, remaining bytes big-endian.
def tar_put_num(p: uint8*, n: sizet, v: uint64) -> int:
if n < 2: return -1
digits: sizet = n - 1
if v <= ((uint64(1) << (3 * digits)) - 1):
i: sizet = digits
while i > 0:
i -= 1
p[i] = uint8(48 + (v & 7))
v = v >> 3
p[digits] = 0
return 0
if n < 9: return -1
p[0] = 128
i2: sizet = n
while i2 > 1:
i2 -= 1
p[i2] = uint8(v & 0xFF)
v = v >> 8
return 0
# Parse up to the first non-digit as a decimal value (a PAX mtime may carry a fraction).
def tar_dec(v: uint8*, n: sizet) -> uint64:
out: uint64 = 0
for i in range(int(n)):
c: uint8 = v[i]
if c < '0' or c > '9': break
out = out * 10 + uint64(c - '0')
return out
def tar_digits(v: sizet) -> sizet:
n: sizet = 1
while v >= 10:
v = v // 10
n += 1
return n
def tar_put_digits(buf: uint8*, v: sizet) -> sizet:
n: sizet = tar_digits(v)
i: sizet = n
while i > 0:
i -= 1
buf[i] = uint8(48 + (v % 10))
v = v // 10
return n
# Length of a NUL-terminated string inside a fixed-size header field.
def tar_field_len(p: uint8*, n: sizet) -> sizet:
i: sizet = 0
while i < n and p[i] != 0: i += 1
return i
def tar_pad(n: uint64) -> sizet:
return sizet(((n + TAR_BLOCK - 1) // TAR_BLOCK) * TAR_BLOCK)
# Header checksum: the unsigned byte sum with the checksum field itself read as eight spaces.
def tar_checksum(block: uint8*) -> uint32:
sum: uint32 = 0
for i in range(int(TAR_BLOCK)):
if i >= TAR_OFF_CHKSUM and i < TAR_OFF_CHKSUM + 8: sum += 32
else: sum += uint32(block[i])
return sum
def tar_is_zero(block: uint8*) -> bool:
for i in range(int(TAR_BLOCK)):
if block[i] != 0: return False
return True
# ============================================================
# memory helpers
# ============================================================
def tar_dup(p: uint8*, n: sizet) -> bytes:
buf: bytes = bytes(stdlib.malloc(n + 1))
if buf is None: return None
if n > 0: string.memcpy(buf, p, n)
buf[n] = 0
return buf
def tar_dup_str(src: bytes) -> bytes:
if src is None: return None
return tar_dup(uint8*(src), string.strlen(src))
def tar_key_is(key: uint8*, klen: sizet, name: str) -> bool:
if klen != string.strlen(name): return False
return string.memcmp(key, name, klen) == 0
# Split a member path into the ustar name/prefix fields: `*name_off` is the start of the part kept
# in the name field, `*prefix_len` the length of the part kept in the prefix field. -1 when the path
# does not fit; the empty part is reported as offset 0 with prefix length 0.
def tar_split_name(name: bytes, n: sizet, name_off: sizet*, prefix_len: sizet*) -> int:
*name_off = 0
*prefix_len = 0
if n <= TAR_NAME_MAX: return 0
i: sizet = n
while i > 0:
i -= 1
if name[i] != '/': continue
head: sizet = i
tail: sizet = n - i - 1
if head == 0 or tail == 0: continue
if head <= TAR_PREFIX_MAX and tail <= TAR_NAME_MAX:
*name_off = i + 1
*prefix_len = head
return 0
return -1
# ============================================================
# TarInfo / TarPax
# ============================================================
struct TarInfo:
name: bytes
name_len: sizet
linkname: bytes
linkname_len: sizet
uname: bytes
gname: bytes
mode: uint32
uid: uint64
gid: uint64
size: uint64
mtime: uint64
typeflag: char
devmajor: uint32
devminor: uint32
data_off: uint64
next: TarInfo*
# PAX overrides carried by an 'x' header (next member only) or a 'g' header (all following members).
struct TarPax:
path: bytes
linkpath: bytes
uname: bytes
gname: bytes
size: uint64
mtime: uint64
uid: uint64
gid: uint64
mask: int
def tar_pax_new() -> TarPax*:
return TarPax*(stdlib.calloc(1, TarPax.__sizeof__()))
def tar_pax_clear(px: TarPax*):
if px.path is not None: stdlib.free(px.path)
if px.linkpath is not None: stdlib.free(px.linkpath)
if px.uname is not None: stdlib.free(px.uname)
if px.gname is not None: stdlib.free(px.gname)
px.path = None
px.linkpath = None
px.uname = None
px.gname = None
px.size = 0
px.mtime = 0
px.uid = 0
px.gid = 0
px.mask = 0
def tar_pax_free(px: TarPax*):
if px is None: return
tar_pax_clear(px)
stdlib.free(px)
# Store one key/value pair. Keys with no representation here (atime, ctime, comment, ...) are read
# and dropped.
def tar_pax_set(px: TarPax*, key: uint8*, klen: sizet, val: uint8*, vlen: sizet):
if tar_key_is(key, klen, "path"):
if px.path is not None: stdlib.free(px.path)
px.path = tar_dup(val, vlen)
px.mask = px.mask | PAX_PATH
elif tar_key_is(key, klen, "linkpath"):
if px.linkpath is not None: stdlib.free(px.linkpath)
px.linkpath = tar_dup(val, vlen)
px.mask = px.mask | PAX_LINKPATH
elif tar_key_is(key, klen, "uname"):
if px.uname is not None: stdlib.free(px.uname)
px.uname = tar_dup(val, vlen)
px.mask = px.mask | PAX_UNAME
elif tar_key_is(key, klen, "gname"):
if px.gname is not None: stdlib.free(px.gname)
px.gname = tar_dup(val, vlen)
px.mask = px.mask | PAX_GNAME
elif tar_key_is(key, klen, "size"):
px.size = tar_dec(val, vlen)
px.mask = px.mask | PAX_SIZE
elif tar_key_is(key, klen, "mtime"):
px.mtime = tar_dec(val, vlen)
px.mask = px.mask | PAX_MTIME
elif tar_key_is(key, klen, "uid"):
px.uid = tar_dec(val, vlen)
px.mask = px.mask | PAX_UID
elif tar_key_is(key, klen, "gid"):
px.gid = tar_dec(val, vlen)
px.mask = px.mask | PAX_GID
# Parse a PAX payload: records are "<len> <key>=<value>\n", where <len> covers the whole record.
def tar_pax_parse(px: TarPax*, p: uint8*, n: sizet) -> int:
cur: sizet = 0
while cur < n:
start: sizet = cur
ln: sizet = 0
while cur < n and p[cur] >= '0' and p[cur] <= '9':
ln = ln * 10 + sizet(p[cur] - '0')
cur += 1
if cur >= n or p[cur] != ' ': return -1
cur += 1
if ln == 0 or start + ln > n: return -1
rec_end: sizet = start + ln
ks: sizet = cur
while cur < rec_end and p[cur] != '=': cur += 1
if cur >= rec_end: return -1
klen: sizet = cur - ks
cur += 1
vs: sizet = cur
while cur < rec_end and p[cur] != '\n': cur += 1
if cur < rec_end: tar_pax_set(px, p + ks, klen, p + vs, cur - vs)
cur = rec_end
return 0
# Append one "<len> <key>=<value>\n" record; returns the new write offset.
def tar_pax_record(buf: uint8*, off: sizet, key: str, val: bytes, val_len: sizet) -> sizet:
klen: sizet = string.strlen(key)
body: sizet = 1 + klen + 1 + val_len + 1
total: sizet = body + 1
while True:
nd: sizet = tar_digits(total)
if nd + body == total: break
total = nd + body
p: sizet = off + tar_put_digits(buf + off, total)
buf[p] = ' '
p += 1
string.memcpy(buf + p, key, klen)
p += klen
buf[p] = '='
p += 1
if val_len > 0: string.memcpy(buf + p, val, val_len)
p += val_len
buf[p] = '\n'
p += 1
return p
# ============================================================
# raw file IO
# ============================================================
def tar_write_all(h: wb.HANDLE, p: uint8*, n: sizet) -> int:
done: sizet = 0
while done < n:
chunk: sizet = n - done
if chunk > 0xFFFFFF00: chunk = 0xFFFFFF00
written: ulong = 0
if wf.WriteFile(h, p + done, ulong(chunk), &written, None) == 0 or written == 0:
tar_set_error(TAR_ERR_IO, "archive write failed")
return TAR_ERR_IO
done += sizet(written)
return TAR_OK
def tar_read_whole(path: str, out_len: sizet*) -> bytes:
h: wb.HANDLE = wf.CreateFileA(path, wf.GENERIC_READ, ulong(TAR_SHARE), None, wf.OPEN_EXISTING, wf.FILE_ATTRIBUTE_NORMAL, None)
if h == wb.INVALID_HANDLE_VALUE:
tar_set_error(TAR_ERR_OPEN, "cannot open archive for reading")
return None
li: wb.LARGE_INTEGER = wb.LARGE_INTEGER()
if wf.GetFileSizeEx(h, &li) == 0:
wb.CloseHandle(h)
tar_set_error(TAR_ERR_IO, "cannot read the archive size")
return None
sz: sizet = sizet(li.QuadPart)
if sz == 0:
wb.CloseHandle(h)
tar_set_error(TAR_ERR_FORMAT, "an empty file is not a tar archive")
return None
buf: bytes = bytes(stdlib.malloc(sz))
if buf is None:
wb.CloseHandle(h)
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return None
done: sizet = 0
while done < sz:
chunk: sizet = sz - done
if chunk > 0xFFFFFF00: chunk = 0xFFFFFF00
got: ulong = 0
if wf.ReadFile(h, buf + done, ulong(chunk), &got, None) == 0 or got == 0:
wb.CloseHandle(h)
stdlib.free(buf)
tar_set_error(TAR_ERR_IO, "archive read failed")
return None
done += sizet(got)
wb.CloseHandle(h)
*out_len = sz
return buf
# ============================================================
# extraction paths
# ============================================================
# Reject absolute paths, drive-relative paths and any ".." component, so a hostile archive cannot
# escape the destination directory.
def tar_name_safe(name: bytes, n: sizet) -> bool:
if n == 0: return False
if name[0] == '/' or name[0] == '\\': return False
if n >= 2 and name[1] == ':': return False
i: sizet = 0
while i + 1 < n:
if name[i] == '.' and name[i + 1] == '.':
head_ok: bool = (i == 0) or (name[i - 1] == '/') or (name[i - 1] == '\\')
tail_ok: bool = (i + 2 >= n) or (name[i + 2] == '/') or (name[i + 2] == '\\')
if head_ok and tail_ok: return False
i += 1
return True
# Join `dest` and a member path into a fresh buffer, dropping a leading "./".
def tar_join(dest: str, name: bytes, name_len: sizet) -> bytes:
dlen: sizet = string.strlen(dest)
sep: sizet = 0
if dlen > 0 and dest[dlen - 1] != '/' and dest[dlen - 1] != '\\': sep = 1
if name_len >= 2 and name[0] == '.' and name[1] == '/':
name = name + 2
name_len = name_len - 2
out: bytes = bytes(stdlib.malloc(dlen + sep + name_len + 1))
if out is None: return None
if dlen > 0: string.memcpy(out, dest, dlen)
if sep != 0:
out[dlen] = '/'
if name_len > 0: string.memcpy(out + dlen + sep, name, name_len)
out[dlen + sep + name_len] = 0
return out
# Create a directory and every missing parent; an existing component is not an error.
def tar_mkdir_p(path: bytes) -> int:
n: sizet = string.strlen(path)
if n == 0: return -1
for i in range(int(n)):
c: char = path[i]
if c != '/' and c != '\\': continue
path[i] = 0
if i > 0: wf.CreateDirectoryA(path, None)
path[i] = '/'
wf.CreateDirectoryA(path, None)
return 0
def tar_mkdir_parent(path: bytes) -> int:
n: sizet = string.strlen(path)
i: sizet = n
while i > 0:
i -= 1
if path[i] == '/' or path[i] == '\\':
path[i] = 0
tar_mkdir_p(path)
path[i] = '/'
return 0
return 0
# ============================================================
# TarFile
# ============================================================
class TarFile:
path: str
mode: int = -1 # 0 = read, 1 = write
fmt: int = TAR_FORMAT_PAX
ok: int = 0
closed: int = 0
data: bytes = None # read mode: whole archive in memory
data_len: sizet = 0
head: TarInfo* = None
tail: TarInfo* = None
count: sizet = 0
gpx: TarPax* = None # PAX global overrides
lpx: TarPax* = None # PAX overrides for the next member
pending_name: bytes = None # GNU 'L'
pending_link: bytes = None # GNU 'K'
fh: wb.HANDLE = None # write mode handle
extracted: sizet = 0
skipped: sizet = 0
def __init__(self, path: str, mode: str = "r", fmt: int = TAR_FORMAT_PAX):
self.path = path
self.fmt = fmt
self.gpx = tar_pax_new()
self.lpx = tar_pax_new()
if self.gpx is None or self.lpx is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
self.closed = 1
return
if string.strcmp(mode, "r") == 0: self.mode = 0
elif string.strcmp(mode, "w") == 0: self.mode = 1
else:
tar_set_error(TAR_ERR_MODE, "mode must be \"r\" or \"w\"")
self.closed = 1
return
if self.mode == 0:
self.data = tar_read_whole(path, &self.data_len)
if self.data is None:
self.closed = 1
return
if self._parse() != TAR_OK:
stdlib.free(self.data)
self.data = None
self.data_len = 0
self._clear_infos()
self.closed = 1
return
else:
self.fh = wf.CreateFileA(path, wf.GENERIC_WRITE, ulong(TAR_SHARE), None, wf.CREATE_ALWAYS, wf.FILE_ATTRIBUTE_NORMAL, None)
if self.fh == wb.INVALID_HANDLE_VALUE:
tar_set_error(TAR_ERR_OPEN, "cannot create archive for writing")
self.fh = None
self.closed = 1
return
self.ok = 1
# ---- lifecycle ----
def _clear_infos(self):
m: TarInfo* = self.head
while m is not None:
nxt: TarInfo* = m.next
if m.name is not None: stdlib.free(m.name)
if m.linkname is not None: stdlib.free(m.linkname)
if m.uname is not None: stdlib.free(m.uname)
if m.gname is not None: stdlib.free(m.gname)
stdlib.free(m)
m = nxt
self.head = None
self.tail = None
self.count = 0
def close(self):
if self.closed != 0: return
if self.mode == 0:
if self.data is not None: stdlib.free(self.data)
self.data = None
self.data_len = 0
self._clear_infos()
else:
if self.fh is not None:
self._write_zero_blocks(TAR_BLOCK * 2) # end-of-archive marker
wb.CloseHandle(self.fh)
self.fh = None
if self.pending_name is not None:
stdlib.free(self.pending_name)
self.pending_name = None
if self.pending_link is not None:
stdlib.free(self.pending_link)
self.pending_link = None
tar_pax_free(self.gpx)
tar_pax_free(self.lpx)
self.gpx = None
self.lpx = None
self.closed = 1
self.ok = 0
# ---- read mode ----
def _append_info(self) -> TarInfo*:
node: TarInfo* = TarInfo*(stdlib.calloc(1, TarInfo.__sizeof__()))
if node is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return None
if self.tail is None: self.head = node
else: self.tail.next = node
self.tail = node
self.count += 1
return node
def _parse(self) -> int:
if self.data_len < TAR_BLOCK:
tar_set_error(TAR_ERR_FORMAT, "smaller than one tar block")
return TAR_ERR_FORMAT
d: bytes = self.data
pos: sizet = 0
while pos + TAR_BLOCK <= self.data_len:
blk: uint8* = uint8*(d + pos)
if tar_is_zero(blk): return TAR_OK
if tar_checksum(blk) != uint32(tar_num(blk + TAR_OFF_CHKSUM, 8)):
tar_set_error(TAR_ERR_CHECKSUM, "header checksum mismatch")
return TAR_ERR_CHECKSUM
typeflag: char = char(blk[TAR_OFF_TYPEFLAG])
hdr_size: uint64 = tar_num(blk + TAR_OFF_SIZE, 12)
payload: sizet = pos + TAR_BLOCK
if hdr_size > uint64(self.data_len - payload):
tar_set_error(TAR_ERR_FORMAT, "member payload is truncated")
return TAR_ERR_FORMAT
if typeflag == TAR_GNU_LONGNAME or typeflag == TAR_GNU_LONGLINK:
slot: bytes* = &self.pending_name
if typeflag == TAR_GNU_LONGLINK: slot = &self.pending_link
if self._store_pending(slot, d, payload, sizet(hdr_size)) != TAR_OK: return TAR_ERR_MEMORY
pos = payload + tar_pad(hdr_size)
continue
if typeflag == TAR_PAX_NEXT or typeflag == TAR_PAX_GLOBAL:
px: TarPax* = self.lpx
if typeflag == TAR_PAX_GLOBAL: px = self.gpx
else: tar_pax_clear(self.lpx)
if tar_pax_parse(px, uint8*(d + payload), sizet(hdr_size)) != 0:
tar_set_error(TAR_ERR_FORMAT, "bad PAX extended header")
return TAR_ERR_FORMAT
pos = payload + tar_pad(hdr_size)
continue
eff: uint64 = 0
if self._read_entry(blk, typeflag, payload, &eff) != TAR_OK: return tarfile_error_code_val
pos = payload + tar_pad(eff)
return TAR_OK
def _store_pending(self, slot: bytes*, d: bytes, payload: sizet, size: sizet) -> int:
buf: bytes = tar_dup(uint8*(d + payload), size)
if buf is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return TAR_ERR_MEMORY
if *slot is not None: stdlib.free(*slot)
*slot = buf
return TAR_OK
# Build one TarInfo from a header block, applying the pending GNU long names and the PAX
# overrides; `*out_size` receives the payload length to advance by.
def _read_entry(self, blk: uint8*, typeflag: char, payload: sizet, out_size: uint64*) -> int:
info: TarInfo* = self._append_info()
if info is None: return TAR_ERR_MEMORY
lpx: TarPax* = self.lpx
gpx: TarPax* = self.gpx
name_len: sizet = tar_field_len(blk + TAR_OFF_NAME, TAR_NAME_MAX)
prefix_len: sizet = tar_field_len(blk + TAR_OFF_PREFIX, TAR_PREFIX_MAX)
if prefix_len > 0:
info.name = bytes(stdlib.malloc(prefix_len + 1 + name_len + 1))
if info.name is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return TAR_ERR_MEMORY
string.memcpy(info.name, blk + TAR_OFF_PREFIX, prefix_len)
info.name[prefix_len] = '/'
if name_len > 0: string.memcpy(info.name + prefix_len + 1, blk + TAR_OFF_NAME, name_len)
info.name[prefix_len + 1 + name_len] = 0
info.name_len = prefix_len + 1 + name_len
else:
info.name = tar_dup(blk + TAR_OFF_NAME, name_len)
info.name_len = name_len
if info.name is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return TAR_ERR_MEMORY
info.linkname = tar_dup(blk + TAR_OFF_LINKNAME, tar_field_len(blk + TAR_OFF_LINKNAME, TAR_LINKNAME_MAX))
if info.linkname is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return TAR_ERR_MEMORY
info.linkname_len = string.strlen(info.linkname)
info.uname = tar_dup(blk + TAR_OFF_UNAME, tar_field_len(blk + TAR_OFF_UNAME, 32))
info.gname = tar_dup(blk + TAR_OFF_GNAME, tar_field_len(blk + TAR_OFF_GNAME, 32))
if self.pending_name is not None:
stdlib.free(info.name)
info.name = self.pending_name
info.name_len = string.strlen(info.name)
self.pending_name = None
if self.pending_link is not None:
stdlib.free(info.linkname)
info.linkname = self.pending_link
info.linkname_len = string.strlen(info.linkname)
self.pending_link = None
pv: bytes = tar_pax_str(lpx, gpx, PAX_PATH)
if pv is not None:
dup: bytes = tar_dup_str(pv)
if dup is not None:
stdlib.free(info.name)
info.name = dup
info.name_len = string.strlen(dup)
lv: bytes = tar_pax_str(lpx, gpx, PAX_LINKPATH)
if lv is not None:
dup2: bytes = tar_dup_str(lv)
if dup2 is not None:
stdlib.free(info.linkname)
info.linkname = dup2
info.linkname_len = string.strlen(dup2)
uv: bytes = tar_pax_str(lpx, gpx, PAX_UNAME)
if uv is not None: tar_replace_str(&info.uname, uv)
gv: bytes = tar_pax_str(lpx, gpx, PAX_GNAME)
if gv is not None: tar_replace_str(&info.gname, gv)
info.typeflag = typeflag
info.mode = uint32(tar_num(blk + TAR_OFF_MODE, 8))
info.uid = tar_num(blk + TAR_OFF_UID, 8)
info.gid = tar_num(blk + TAR_OFF_GID, 8)
info.size = tar_num(blk + TAR_OFF_SIZE, 12)
info.mtime = tar_num(blk + TAR_OFF_MTIME, 12)
info.devmajor = uint32(tar_num(blk + TAR_OFF_DEVMAJOR, 8))
info.devminor = uint32(tar_num(blk + TAR_OFF_DEVMINOR, 8))
if (lpx.mask & PAX_SIZE) != 0: info.size = lpx.size
elif (gpx.mask & PAX_SIZE) != 0: info.size = gpx.size
if (lpx.mask & PAX_MTIME) != 0: info.mtime = lpx.mtime
elif (gpx.mask & PAX_MTIME) != 0: info.mtime = gpx.mtime
if (lpx.mask & PAX_UID) != 0: info.uid = lpx.uid
elif (gpx.mask & PAX_UID) != 0: info.uid = gpx.uid
if (lpx.mask & PAX_GID) != 0: info.gid = lpx.gid
elif (gpx.mask & PAX_GID) != 0: info.gid = gpx.gid
info.data_off = uint64(payload)
*out_size = info.size
tar_pax_clear(lpx)
return TAR_OK
# ---- read accessors ----
def first(self) -> TarInfo*:
return self.head
def find(self, name: str) -> TarInfo*:
m: TarInfo* = self.head
while m is not None:
if string.strcmp(m.name, name) == 0: return m
m = m.next
return None
def is_file(self, m: TarInfo*) -> bool:
if m is None: return False
return m.typeflag == TAR_REG or m.typeflag == TAR_REG_OLD or m.typeflag == TAR_CONT
def read_member(self, m: TarInfo*, out_len: sizet*) -> bytes:
if self.ok == 0 or self.mode != 0 or not self.is_file(m): return None
n: sizet = sizet(m.size)
buf: bytes = bytes(stdlib.malloc(n + 1))
if buf is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return None
if n > 0: string.memcpy(buf, self.data + sizet(m.data_off), n)
buf[n] = 0
*out_len = n
return buf
# Write every regular file and directory member under `dest`; other types are counted in
# `skipped` and left to the caller.
def extractall(self, dest: str) -> int:
if self.ok == 0 or self.mode != 0:
tar_set_error(TAR_ERR_MODE, "extractall needs a reader")
return TAR_ERR_MODE
m: TarInfo* = self.head
while m is not None:
if not tar_name_safe(m.name, m.name_len):
tar_set_error(TAR_ERR_UNSAFE, "member path escapes the destination")
return TAR_ERR_UNSAFE
full: bytes = tar_join(dest, m.name, m.name_len)
if full is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return TAR_ERR_MEMORY
if m.typeflag == TAR_DIR:
tar_mkdir_p(full)
self.extracted += 1
elif self.is_file(m):
tar_mkdir_parent(full)
rc: int = self._write_member(full, m)
if rc != TAR_OK:
stdlib.free(full)
return rc
self.extracted += 1
else:
self.skipped += 1
stdlib.free(full)
m = m.next
return TAR_OK
def _write_member(self, full: bytes, m: TarInfo*) -> int:
h: wb.HANDLE = wf.CreateFileA(full, wf.GENERIC_WRITE, ulong(TAR_SHARE), None, wf.CREATE_ALWAYS, wf.FILE_ATTRIBUTE_NORMAL, None)
if h == wb.INVALID_HANDLE_VALUE:
tar_set_error(TAR_ERR_IO, "cannot create the extracted file")
return TAR_ERR_IO
n: sizet = sizet(m.size)
rc: int = TAR_OK
if n > 0: rc = tar_write_all(h, uint8*(self.data + sizet(m.data_off)), n)
wb.CloseHandle(h)
return rc
# ---- write mode ----
def _check_writer(self) -> int:
if self.ok == 0:
tar_set_error(TAR_ERR_CLOSED, "archive is not open")
return TAR_ERR_CLOSED
if self.mode != 1:
tar_set_error(TAR_ERR_MODE, "this operation needs a writer")
return TAR_ERR_MODE
return TAR_OK
def _write_zero_blocks(self, total: sizet) -> int:
blk: uint8* = uint8*(stdlib.malloc(TAR_BLOCK))
if blk is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return TAR_ERR_MEMORY
string.memset(blk, 0, TAR_BLOCK)
written: sizet = 0
rc: int = TAR_OK
while written < total:
rc = tar_write_all(self.fh, blk, TAR_BLOCK)
if rc != TAR_OK: break
written += TAR_BLOCK
stdlib.free(blk)
return rc
def _write_payload(self, data: bytes, n: sizet) -> int:
rc: int = TAR_OK
if n > 0: rc = tar_write_all(self.fh, uint8*(data), n)
if rc != TAR_OK: return rc
pad: sizet = tar_pad(uint64(n)) - n
if pad > 0: return self._write_zero_blocks(pad)
return TAR_OK
# Build and write one ustar header block. `name_len` bytes of `name` starting at `name_off`
# fill the name field, and `prefix_len` bytes from the start fill the prefix field.
def _write_ustar(self, name: bytes, name_off: sizet, name_len: sizet, prefix_len: sizet, mode: uint32, uid: uint64, gid: uint64, size: uint64, mtime: uint64, typeflag: char, link: bytes, link_len: sizet) -> int:
blk: uint8* = uint8*(stdlib.malloc(TAR_BLOCK))
if blk is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return TAR_ERR_MEMORY
string.memset(blk, 0, TAR_BLOCK)
if name_len > 0: string.memcpy(blk + TAR_OFF_NAME, name + name_off, name_len)
tar_put_num(blk + TAR_OFF_MODE, 8, uint64(mode))
tar_put_num(blk + TAR_OFF_UID, 8, uid)
tar_put_num(blk + TAR_OFF_GID, 8, gid)
tar_put_num(blk + TAR_OFF_SIZE, 12, size)
tar_put_num(blk + TAR_OFF_MTIME, 12, mtime)
string.memset(blk + TAR_OFF_CHKSUM, 32, 8)
blk[TAR_OFF_TYPEFLAG] = uint8(typeflag)
if link_len > 0: string.memcpy(blk + TAR_OFF_LINKNAME, link, link_len)
string.memcpy(blk + TAR_OFF_MAGIC, "ustar", 5)
blk[TAR_OFF_MAGIC + 5] = 0
blk[TAR_OFF_MAGIC + 6] = '0'
blk[TAR_OFF_MAGIC + 7] = '0'
if prefix_len > 0: string.memcpy(blk + TAR_OFF_PREFIX, name, prefix_len)
tar_put_num(blk + TAR_OFF_CHKSUM, 7, uint64(tar_checksum(blk)))
blk[TAR_OFF_CHKSUM + 7] = ' '
rc: int = tar_write_all(self.fh, blk, TAR_BLOCK)
stdlib.free(blk)
return rc
# Emit an 'x' header carrying the overrides that do not fit a ustar header. `tag` names the 'x'
# member itself; `rec_name`/`rec_link` are the records actually emitted (each optional).
def _write_pax(self, tag: bytes, tag_len: sizet, rec_name: bytes, rec_name_len: sizet, rec_link: bytes, rec_link_len: sizet, mtime: uint64) -> int:
cap: sizet = TAR_BLOCK + rec_name_len + rec_link_len + 64
buf: uint8* = uint8*(stdlib.malloc(cap))
if buf is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return TAR_ERR_MEMORY
off: sizet = 0
if rec_name_len > 0: off = tar_pax_record(buf, off, "path", rec_name, rec_name_len)
if rec_link_len > 0: off = tar_pax_record(buf, off, "linkpath", rec_link, rec_link_len)
tagdup: bytes = tar_dup(uint8*(tag), tag_len)
if tagdup is None:
stdlib.free(buf)
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return TAR_ERR_MEMORY
rc: int = self._write_ustar(tagdup, 0, tag_len, 0, TAR_MODE_FILE, 0, 0, uint64(off), mtime, TAR_PAX_NEXT, None, 0)
if rc == TAR_OK: rc = self._write_payload(bytes(buf), off)
stdlib.free(tagdup)
stdlib.free(buf)
return rc
# Emit the header of one member, preceded by a PAX header when the path or the link target is
# too long for the ustar fields.
def _write_entry(self, name: str, name_len: sizet, mode: uint32, uid: uint64, gid: uint64, size: uint64, mtime: uint64, typeflag: char, link: str, link_len: sizet) -> int:
if name_len == 0:
tar_set_error(TAR_ERR_FORMAT, "empty member name")
return TAR_ERR_FORMAT
name_off: sizet = 0
prefix_len: sizet = 0
name_fits: bool = tar_split_name(uint8*(name), name_len, &name_off, &prefix_len) == 0
link_fits: bool = link_len <= TAR_LINKNAME_MAX
if not name_fits or not link_fits:
if self.fmt != TAR_FORMAT_PAX:
tar_set_error(TAR_ERR_UNSUPPORTED, "path does not fit ustar: use the PAX format")
return TAR_ERR_UNSUPPORTED
tag_len: sizet = name_len
if tag_len > TAR_NAME_MAX: tag_len = TAR_NAME_MAX
rec_name: bytes = None
rec_name_len: sizet = 0
if not name_fits:
rec_name = name
rec_name_len = name_len
name_off = 0
prefix_len = 0
name_len = TAR_NAME_MAX
rec_link: bytes = None
rec_link_len: sizet = 0
if not link_fits:
rec_link = link
rec_link_len = link_len
link_len = TAR_LINKNAME_MAX
rc: int = self._write_pax(name, tag_len, rec_name, rec_name_len, rec_link, rec_link_len, mtime)
if rc != TAR_OK: return rc
return self._write_ustar(uint8*(name), name_off, name_len, prefix_len, mode, uid, gid, size, mtime, typeflag, uint8*(link), link_len)
# ---- public writers ----
def add_file(self, name: str, data: bytes, data_len: sizet, mode: uint32 = TAR_MODE_FILE, mtime: uint64 = 0) -> int:
if self._check_writer() != TAR_OK: return tarfile_error_code_val
rc: int = self._write_entry(name, string.strlen(name), mode, 0, 0, uint64(data_len), mtime, TAR_REG, None, 0)
if rc != TAR_OK: return rc
return self._write_payload(data, data_len)
def add_dir(self, name: str, mode: uint32 = TAR_MODE_DIR, mtime: uint64 = 0) -> int:
if self._check_writer() != TAR_OK: return tarfile_error_code_val
n: sizet = string.strlen(name)
if n == 0:
tar_set_error(TAR_ERR_FORMAT, "empty member name")
return TAR_ERR_FORMAT
if name[n - 1] == '/':
return self._write_entry(name, n, mode, 0, 0, uint64(0), mtime, TAR_DIR, None, 0)
buf: bytes = bytes(stdlib.malloc(n + 2))
if buf is None:
tar_set_error(TAR_ERR_MEMORY, "out of memory")
return TAR_ERR_MEMORY
string.memcpy(buf, name, n)
buf[n] = '/'
buf[n + 1] = 0
rc: int = int(self._write_entry(buf, n + 1, mode, 0, 0, uint64(0), mtime, TAR_DIR, None, 0))
stdlib.free(buf)
return rc
def add_symlink(self, name: str, target: str, mtime: uint64 = 0) -> int:
if self._check_writer() != TAR_OK: return tarfile_error_code_val
return self._write_entry(name, string.strlen(name), TAR_MODE_LINK, 0, 0, uint64(0), mtime, TAR_SYMLINK, target, string.strlen(target))
def add_hardlink(self, name: str, target: str, mtime: uint64 = 0) -> int:
if self._check_writer() != TAR_OK: return tarfile_error_code_val
return self._write_entry(name, string.strlen(name), TAR_MODE_FILE, 0, 0, uint64(0), mtime, TAR_HARDLINK, target, string.strlen(target))
# ============================================================
# PAX string lookup / replacement
# ============================================================
def tar_pax_str(lpx: TarPax*, gpx: TarPax*, bit: int) -> bytes:
if (lpx.mask & bit) != 0:
if bit == PAX_PATH: return lpx.path
if bit == PAX_LINKPATH: return lpx.linkpath
if bit == PAX_UNAME: return lpx.uname
return lpx.gname
if (gpx.mask & bit) != 0:
if bit == PAX_PATH: return gpx.path
if bit == PAX_LINKPATH: return gpx.linkpath
if bit == PAX_UNAME: return gpx.uname
return gpx.gname
return None
def tar_replace_str(slot: bytes*, src: bytes):
if src is None: return
dup: bytes = tar_dup_str(src)
if dup is None: return
if *slot is not None: stdlib.free(*slot)
*slot = dup
Reference in new issue View Git Blame Copy Permalink