-
Notifications
You must be signed in to change notification settings - Fork 7
/
util.py
236 lines (182 loc) · 6.6 KB
/
util.py
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
# SPDX-License-Identifier: GPL-3.0-or-later
from collections.abc import Iterable
from collections.abc import Iterator
from contextlib import contextmanager
from os import chdir
from os import getcwd
from pathlib import Path
from re import split
from subprocess import STDOUT
from subprocess import CalledProcessError
from subprocess import check_output
from sys import exit
from sys import stderr
from tempfile import mkstemp
from traceback import print_stack
from typing import IO
from typing import AnyStr
from typing import Dict
from typing import List
from typing import Optional
from typing import Set
from typing import Union
from libkeyringctl.types import Fingerprint
from libkeyringctl.types import Trust
@contextmanager
def cwd(new_dir: Path) -> Iterator[None]:
"""Change to a new current working directory in a context and go back to the previous dir after the context is done
Parameters
----------
new_dir: A path to change to
"""
previous_dir = getcwd()
chdir(new_dir)
try:
yield
finally:
chdir(previous_dir)
def natural_sort_path(_list: Iterable[Path]) -> Iterable[Path]:
"""Sort an Iterable of Paths naturally
Parameters
----------
_list: An iterable containing paths to be sorted
Return
------
An Iterable of paths that are naturally sorted
"""
def convert_text_chunk(text: str) -> Union[int, str]:
"""Convert input text to int or str
Parameters
----------
text: An input string
Returns
-------
Either an integer if text is a digit, else text in lower-case representation
"""
return int(text) if text.isdigit() else text.lower()
def alphanum_key(key: Path) -> List[Union[int, str]]:
"""Retrieve an alphanumeric key from a Path, that can be used in sorted()
Parameters
----------
key: A path for which to create a key
Returns
-------
A list of either int or str objects that may serve as 'key' argument for sorted()
"""
return [convert_text_chunk(c) for c in split("([0-9]+)", str(key.name))]
return sorted(_list, key=alphanum_key)
def system(cmd: List[str], _stdin: Optional[IO[AnyStr]] = None, exit_on_error: bool = False) -> str:
"""Execute a command using check_output
Parameters
----------
cmd: A list of strings to be fed to check_output
_stdin: input fd used for the spawned process
exit_on_error: Whether to exit the script when encountering an error (defaults to False)
Raises
------
CalledProcessError: If not exit_on_error and `check_output()` encounters an error
Returns
-------
The output of cmd
"""
try:
return check_output(cmd, stderr=STDOUT, stdin=_stdin).decode()
except CalledProcessError as e:
stderr.buffer.write(bytes(e.stdout, encoding="utf8"))
print_stack()
if exit_on_error:
exit(e.returncode)
raise e
def absolute_path(path: str) -> Path:
"""Return the absolute path of a given str
Parameters
----------
path: A string representing a path
Returns
-------
The absolute path representation of path
"""
return Path(path).absolute()
def transform_fd_to_tmpfile(working_dir: Path, sources: List[Path]) -> None:
"""Transforms an input list of paths from any file descriptor of the current process to a tempfile in working_dir.
Using this function on fd inputs allow to pass the content to another process while hidepid is active and /proc
not visible for the other process.
Parameters
----------
working_dir: A directory to use for temporary files
sources: Paths that should be iterated and all fd's transformed to tmpfiles
"""
for index, source in enumerate(sources):
if str(source).startswith("/proc/self/fd"):
file = mkstemp(dir=working_dir, prefix=f"{source.name}", suffix=".fd")[1]
with open(file, mode="wb") as f:
f.write(source.read_bytes())
f.flush()
sources[index] = Path(file)
def get_cert_paths(paths: Iterable[Path]) -> Set[Path]:
"""Walks a list of paths and resolves all discovered certificate paths
Parameters
----------
paths: A list of paths to walk and resolve to certificate paths.
Returns
-------
A set of paths to certificates
"""
# depth first search certificate paths
cert_paths: Set[Path] = set()
visit: List[Path] = list(paths)
while visit:
path = visit.pop()
# this level contains a certificate, abort depth search
if list(path.glob("*.asc")):
cert_paths.add(path)
continue
visit.extend([path for path in path.iterdir() if path.is_dir()])
return cert_paths
def get_parent_cert_paths(paths: Iterable[Path]) -> Set[Path]:
"""Walks a list of paths upwards and resolves all discovered parent certificate paths
Parameters
----------
paths: A list of paths to walk and resolve to certificate paths.
Returns
-------
A set of paths to certificates
"""
# depth first search certificate paths
cert_paths: Set[Path] = set()
visit: List[Path] = list(paths)
while visit:
node = visit.pop().parent
# this level contains a certificate, abort depth search
if "keyring" == node.parent.parent.parent.name:
cert_paths.add(node)
continue
visit.append(node)
return cert_paths
def contains_fingerprint(fingerprints: Iterable[Fingerprint], fingerprint: Fingerprint) -> bool:
"""Returns weather an iterable structure of fingerprints contains a specific fingerprint
Parameters
----------
fingerprints: Iteratable structure of fingerprints that should be searched
fingerprint: Fingerprint to search for
Returns
-------
Weather an iterable structure of fingerprints contains a specific fingerprint
"""
return any(filter(lambda e: str(e).endswith(fingerprint), fingerprints))
def filter_fingerprints_by_trust(trusts: Dict[Fingerprint, Trust], trust: Trust) -> List[Fingerprint]:
"""Filters a dict of Fingerprint to Trust by a passed Trust parameter and returns the matching fingerprints.
Parameters
----------
trusts: Dict of Fingerprint to Trust that should be filtered based on the trust parameter
trust: Trust that should be used to filter the trusts dict
Returns
-------
The matching fingerprints of the dict filtered by trust
"""
return list(
map(
lambda item: item[0],
filter(lambda item: trust == item[1], trusts.items()),
)
)