python-libarchive/libarchive/zip.py at 0fe09307f1504c471f94dd7f882c565597f066c7 · smartfile/python-libarchive · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
import os
import time
from libarchive import is_archive, Entry, SeekableArchive, _libarchive
from zipfile import ZIP_STORED, ZIP_DEFLATED


def is_zipfile(filename):
    return is_archive(filename, formats=('zip',))


def sanitize_filename(filename, base_path=os.getcwd()):
    abs_path = os.path.abspath(os.path.join(base_path, filename))
    if not abs_path.startswith(os.path.abspath(base_path) + os.sep):
        raise ValueError("Invalid filename: Potential directory traversal attempt detected.")
    return os.path.basename(abs_path)  # Ensures only filename is extracted

class ZipEntry(Entry):
    def __init__(self, *args, **kwargs):
        super(ZipEntry, self).__init__(*args, **kwargs)

    def get_filename(self):
        return self.pathname

    def set_filename(self, value):
        self.pathname = value

    filename = property(get_filename, set_filename)

    def get_file_size(self):
        return self.size

    def set_file_size(self, value):
        assert isinstance(value, int), 'Please provide size as int or long.'
        self.size = value

    file_size = property(get_file_size, set_file_size)

    def get_date_time(self):
        return time.localtime(self.mtime)[0:6]

    def set_date_time(self, value):
        assert isinstance(value, tuple), 'mtime should be tuple (year, month, day, hour, minute, second).'
        assert len(value) == 6, 'mtime should be tuple (year, month, day, hour, minute, second).'
        self.mtime = time.mktime(value + (0, 0, 0))

    date_time = property(get_date_time, set_date_time)

    header_offset = Entry.header_position

    def _get_missing(self):
        raise NotImplemented()

    def _set_missing(self, value):
        raise NotImplemented()

    compress_type = property(_get_missing, _set_missing)
    comment = property(_get_missing, _set_missing)
    extra = property(_get_missing, _set_missing)
    create_system = property(_get_missing, _set_missing)
    create_version = property(_get_missing, _set_missing)
    extract_version = property(_get_missing, _set_missing)
    reserved = property(_get_missing, _set_missing)
    flag_bits = property(_get_missing, _set_missing)
    volume = property(_get_missing, _set_missing)
    internal_attr = property(_get_missing, _set_missing)
    external_attr = property(_get_missing, _set_missing)
    CRC = property(_get_missing, _set_missing)
    compress_size = property(_get_missing, _set_missing)


class ZipFile(SeekableArchive):
    def __init__(self, f, mode='r', compression=ZIP_DEFLATED, allowZip64=False, password=None,
                 encryption=None):
        self.compression = compression
        self.encryption = encryption
        super(ZipFile, self).__init__(
            f, mode=mode, format='zip', entry_class=ZipEntry, encoding='CP437', password=password
        )

    getinfo = SeekableArchive.getentry

    def set_initial_options(self):
        if self.mode == 'w' and self.compression == ZIP_STORED:
            _libarchive.archive_write_set_format_option(self._a, "zip", "compression", "store")

        if self.mode == 'w' and self.password:
            if not self.encryption:
                self.encryption = "traditional"
            _libarchive.archive_write_set_format_option(self._a, "zip", "encryption", self.encryption)

    def namelist(self):
        return list(self.iterpaths())

    def infolist(self):
        return list(self)

    def open(self, name, mode, pwd=None):
        if mode == 'r':
            if pwd:
                self.add_passphrase(pwd)
            return self.readstream(name)
        else:
            return self.writestream(name)

    def extract(self, name, path=None, pwd=None):
        if pwd:
            self.add_passphrase(pwd)
        if not path:
            path = os.getcwd()
        sanitized_name = sanitize_filename(name)
        return self.readpath(sanitized_name, os.path.join(path, sanitized_name))

    def extractall(self, path, names=None, pwd=None):
        if pwd:
            self.add_passphrase(pwd)
        if not names:
            names = self.namelist()
        if names:
            for name in names:
                sanitized_name = sanitize_filename(name, path)
                self.extract(sanitized_name, path)

    def read(self, name, pwd=None):
        if pwd:
            self.add_passphrase(pwd)
        return super(ZipFile, self).read(name)

    def writestr(self, member, data, compress_type=None):
        if compress_type != self.compression and not (compress_type is None):
            raise Exception('Cannot change compression type for individual entries.')
        return self.write(member, data)

    def setpassword(self, pwd):
        return self.set_passphrase(pwd)

    def testzip(self):
        raise NotImplemented()

    def _get_missing(self):
        raise NotImplemented()

    def _set_missing(self, value):
        raise NotImplemented()

    comment = property(_get_missing, _set_missing)