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
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
|
"""Addon functionality shared by multiple checkers."""
from collections import defaultdict
from functools import partial
from itertools import chain, filterfalse
from pkgcore.ebuild import misc
from pkgcore.ebuild import profiles as profiles_mod
from pkgcore.restrictions import packages
from snakeoil.cli import arghparse
from snakeoil.sequences import iflatten_instance
from snakeoil.strings import pluralism
from .. import base, results
from ..base import PkgcheckUserException
from ..log import logger
from . import caches
class ArchesArgs(arghparse.CommaSeparatedNegations):
"""Parse arches args for the ArchesAddon."""
def __call__(self, parser, namespace, values, option_string=None):
disabled, enabled = self.parse_values(values)
all_arches = namespace.target_repo.known_arches
if not enabled:
# enable all non-prefix arches
enabled = set(arch for arch in all_arches if "-" not in arch)
arches = set(enabled).difference(disabled)
if all_arches and (unknown_arches := arches.difference(all_arches)):
es = pluralism(unknown_arches, plural="es")
unknown = ", ".join(unknown_arches)
valid = ", ".join(sorted(all_arches))
parser.error(f"unknown arch{es}: {unknown} (valid arches: {valid})")
# check if any selected arch only has experimental profiles
for arch in arches:
if all(p.status == "exp" for p in namespace.target_repo.profiles if p.arch == arch):
namespace.exp_profiles_required = True
break
arches = frozenset(arches)
setattr(namespace, self.dest, arches)
namespace.arches = arches
class ArchesAddon(base.Addon):
"""Addon supporting ebuild repository architectures."""
@classmethod
def mangle_argparser(cls, parser):
group = parser.add_argument_group("arches")
group.add_argument(
"-a",
"--arches",
dest="selected_arches",
metavar="ARCH",
default=(),
action=arghparse.Delayed,
target=ArchesArgs,
priority=100,
help="comma separated list of arches to enable/disable",
docs="""
Comma separated list of arches to enable and disable.
To specify disabled arches prefix them with '-'. Note that when
starting the argument list with a disabled arch an equals sign
must be used, e.g. -a=-arch, otherwise the disabled arch
argument is treated as an option.
By default all repo defined arches are used; however,
stable-related checks (e.g. UnstableOnly) default to the set of
arches having stable profiles in the target repo.
""",
)
parser.bind_delayed_default(1000, "arches")(cls._default_arches)
@staticmethod
def _default_arches(namespace, attr):
"""Use all known arches by default."""
setattr(namespace, attr, namespace.target_repo.known_arches)
class KeywordsAddon(base.Addon):
"""Addon supporting various keywords sets."""
def __init__(self, *args):
super().__init__(*args)
special = {"-*"}
self.arches: frozenset[str] = self.options.target_repo.known_arches
unstable = {"~" + x for x in self.arches}
disabled = {"-" + x for x in chain(self.arches, unstable)}
self.valid = special | self.arches | unstable | disabled
# Note: '*' and '~*' are portage-only, i.e. not in the spec, so they
# don't belong in the main tree.
self.portage = {"*", "~*"}
class StableArchesAddon(base.Addon):
"""Addon supporting stable architectures."""
required_addons = (ArchesAddon,)
@classmethod
def mangle_argparser(cls, parser):
parser.bind_delayed_default(1001, "stable_arches")(cls._default_stable_arches)
@staticmethod
def _default_stable_arches(namespace, attr):
"""Determine set of stable arches to use."""
target_repo = namespace.target_repo
if not namespace.selected_arches:
# use known stable arches (GLEP 72) if arches aren't specified
stable_arches = target_repo.config.arches_desc["stable"]
# fallback to determining stable arches from profiles.desc if arches.desc doesn't exist
if not stable_arches:
stable_arches = set().union(
*(repo.profiles.arches("stable") for repo in target_repo.trees)
)
else:
stable_arches = namespace.arches
setattr(namespace, attr, stable_arches)
class UnstatedIuse(results.VersionResult, results.Error):
"""Package is reliant on conditionals that aren't in IUSE."""
def __init__(self, attr, flags, profile=None, num_profiles=None, **kwargs):
super().__init__(**kwargs)
self.attr = attr
self.flags = tuple(flags)
self.profile = profile
self.num_profiles = num_profiles
@property
def desc(self):
msg = [f"attr({self.attr})"]
if self.profile is not None:
if self.num_profiles is not None:
num_profiles = f" ({self.num_profiles} total)"
else:
num_profiles = ""
msg.append(f"profile {self.profile!r}{num_profiles}")
flags = ", ".join(self.flags)
s = pluralism(self.flags)
msg.extend([f"unstated flag{s}", f"[ {flags} ]"])
return ": ".join(msg)
class UseAddon(base.Addon):
"""Addon supporting USE flag functionality."""
def __init__(self, *args):
super().__init__(*args)
target_repo = self.options.target_repo
self.profiles = []
for p in target_repo.profiles:
try:
self.profiles.append(
target_repo.profiles.create_profile(p, load_profile_base=False)
)
except profiles_mod.ProfileError:
continue
# TODO: Figure out if there is a more efficient method to determine a
# repo's global implicit iuse while avoiding profiles cache usage. The
# cache shouldn't be used in order to avoid cache regens when
# performing scanning actions on specific profile files since the
# current ProfilesCheck uses this addon.
if self.profiles:
c_implicit_iuse = set.intersection(*(set(p.iuse_effective) for p in self.profiles))
else:
c_implicit_iuse = set()
known_iuse = set()
known_iuse_expand = set()
for repo in target_repo.trees:
known_iuse.update(flag for matcher, (flag, desc) in repo.config.use_desc)
known_iuse_expand.update(
flag for flags in repo.config.use_expand_desc.values() for flag, desc in flags
)
self.collapsed_iuse = misc.non_incremental_collapsed_restrict_to_data(
((packages.AlwaysTrue, known_iuse),),
((packages.AlwaysTrue, known_iuse_expand),),
)
self.global_iuse = frozenset(known_iuse)
self.global_iuse_expand = frozenset(known_iuse_expand)
self.global_iuse_implicit = frozenset(c_implicit_iuse)
self.ignore = not (c_implicit_iuse or known_iuse or known_iuse_expand)
if self.ignore:
logger.debug(
"disabling use/iuse validity checks since no usable "
"use.desc and use.local.desc were found"
)
def allowed_iuse(self, pkg):
return self.collapsed_iuse.pull_data(pkg).union(pkg.local_use)
def get_filter(self, attr=None):
if self.ignore:
return self.fake_use_validate
if attr is not None:
return partial(self.use_validate, attr=attr)
return self.use_validate
@staticmethod
def fake_use_validate(klasses, pkg, seq, attr=None):
return {k: () for k in iflatten_instance(seq, klasses)}, ()
def _flatten_restricts(self, nodes, skip_filter, stated, unstated, attr, restricts=None):
for node in nodes:
k = node
v = restricts if restricts is not None else []
if isinstance(node, packages.Conditional):
# invert it; get only whats not in pkg.iuse
unstated.update(filterfalse(stated.__contains__, node.restriction.vals))
v.append(node.restriction)
yield from self._flatten_restricts(
iflatten_instance(node.payload, skip_filter),
skip_filter,
stated,
unstated,
attr,
v,
)
continue
elif attr == "required_use":
unstated.update(filterfalse(stated.__contains__, node.vals))
yield k, tuple(v)
def _unstated_iuse(self, pkg, attr, unstated_iuse):
"""Determine if packages use unstated IUSE for a given attribute."""
# determine profiles lacking USE flags
if self.profiles:
profiles_unstated = defaultdict(set)
if attr is not None:
for p in self.profiles:
if profile_unstated := unstated_iuse - p.iuse_effective:
profiles_unstated[tuple(sorted(profile_unstated))].add(p.name)
for unstated, profiles in profiles_unstated.items():
profiles = sorted(profiles)
if self.options.verbosity > 0:
for p in profiles:
yield UnstatedIuse(attr, unstated, p, pkg=pkg)
else:
num_profiles = len(profiles)
yield UnstatedIuse(attr, unstated, profiles[0], num_profiles, pkg=pkg)
elif unstated_iuse:
# Remove global defined implicit USE flags, note that standalone
# repos without profiles will currently lack any implicit IUSE.
unstated_iuse -= self.global_iuse_implicit
if unstated_iuse:
yield UnstatedIuse(attr, sorted(unstated_iuse), pkg=pkg)
def use_validate(self, klasses, pkg, seq, attr=None):
skip_filter = (packages.Conditional,) + klasses
nodes = iflatten_instance(seq, skip_filter)
unstated = set()
vals = dict(
self._flatten_restricts(
nodes, skip_filter, stated=pkg.iuse_stripped, unstated=unstated, attr=attr
)
)
return vals, self._unstated_iuse(pkg, attr, unstated)
class NetAddon(base.Addon):
"""Addon supporting network functionality."""
def __init__(self, *args):
super().__init__(*args)
if self.options.timeout == 0:
# set timeout to 0 to never timeout
self.timeout = None
else:
# default to timing out connections after 5 seconds
self.timeout = self.options.timeout if self.options.timeout is not None else 5
@classmethod
def mangle_argparser(cls, parser):
group = parser.add_argument_group("network")
group.add_argument(
"--timeout", type=float, default="5", help="timeout used for network checks"
)
group.add_argument(
"--user-agent", default="Wget/1.20.3 (linux-gnu)", help="custom user agent spoofing"
)
@property
def session(self):
try:
from .net import Session
return Session(
concurrent=self.options.tasks,
timeout=self.timeout,
user_agent=self.options.user_agent,
)
except ImportError as e:
if e.name == "requests":
raise PkgcheckUserException("network checks require requests to be installed")
raise
def init_addon(cls, options, addons_map=None, **kwargs):
"""Initialize a given addon."""
if addons_map is None:
addons_map = {}
try:
addon = addons_map[cls]
except KeyError:
# initialize and inject all required addons for a given addon's inheritance
# tree as kwargs
required_addons = chain.from_iterable(
x.required_addons for x in cls.__mro__ if issubclass(x, base.Addon)
)
kwargs.update(
{
base.param_name(addon): init_addon(addon, options, addons_map)
for addon in required_addons
}
)
# verify the cache type is enabled
if issubclass(cls, caches.CachedAddon) and not options.cache[cls.cache.type]:
raise caches.CacheDisabled(cls.cache)
addon = addons_map[cls] = cls(options, **kwargs)
# force cache updates
force_cache = getattr(options, "force_cache", False)
if isinstance(addon, caches.CachedAddon):
addon.update_cache(force=force_cache)
return addon
|