3
"""Subscribe uploaders to the unapproved queue to their SRU bugs
5
Author: Robie Basak <robie.basak@canonical.com>
7
Subscriptions are made on the principle that if you signed an upload, you need
8
to follow up on review comments on the SRU team. The SRU team have been having
9
trouble getting response to comments in bugs, noticed that often the sponsor
10
isn't subscribed, and hopes this will help.
12
This is intended to be run from a cronjob or timer. Run with --debug to see
13
what is going on, and additionally with --dry-run to see what would happen.
15
Principle of operation: get all key fingerprints of everyone in
16
~ubuntu-uploaders. Then download the .dsc file from everything in Unapproved
17
targetted at the proposed pockets of stable series and see who signed them.
18
Then subscribe this person to all bugs referenced in Launchpad-Bugs-Fixed in
19
the changes file of those uploads if they aren't subscribed at "Discussion"
22
To stop this script from acting on a particular bug, tag it "bot-stop-nagging".
24
It takes a while to get going; hitting the Launchpad API for all uploaders'
25
keys takes about a minute, then a further minute to fetch all the keys to get
26
their subkey fingerprints from keyserver.ubuntu.com.
28
All Ubuntu uploaders are supposed to be in ~ubuntu-uploaders but this isn't
29
enforced except by the correct action of the DMB. If somebody has uploaded but
30
is not a member of the team, this script will not find their key and therefore
31
will not subscribe them to their bugs.
33
The key 6309259455EFCAA8 also signs uploads and is ~ci-train-bot; it isn't a
34
member of ~ubuntu-uploaders and is therefore also ignored. We don't have a way
35
of identifying the actual sponsor in this case.
47
import launchpadlib.launchpad
49
GPG_OUTPUT_PATTERN = re.compile(
50
r"""gpg: Signature made.*
51
gpg: +using (?:RSA|DSA) key (?P<fingerprint>[0-9A-F]+)
56
def find_specific_url(urls, suffix):
58
if url.endswith(suffix):
63
def get_upload_source_urls(upload):
64
if upload.contains_source:
65
return upload.sourceFileUrls()
66
elif upload.contains_copy:
68
copy_source_archive = upload.copy_source_archive
69
# Trigger a problem ValueError exception now rather than later
70
# This is magic launchpadlib behaviour: accessing an attribute of
71
# copy_source_archive may fail later on an access permission issue
72
# due to lazy loading.
73
getattr(copy_source_archive, "self_link")
74
except ValueError as e:
76
f"Cannot access {upload} copy_source_archive attribute: no permission?"
80
upload.copy_source_archive.getPublishedSources(
81
source_name=upload.package_name,
82
version=upload.package_version,
89
raise RuntimeError(f"Cannot find source for {upload}")
92
def find_dsc_signing_fingerprint(dsc_url):
93
with urllib.request.urlopen(
95
) as dsc_fobj, tempfile.TemporaryDirectory() as tmpdir:
96
result = subprocess.run(
98
input=dsc_fobj.read(),
100
env={"GNUPGHOME": tmpdir},
102
if result.returncode != 2:
103
raise RuntimeError("Unknown exit status from gpg")
104
m = GPG_OUTPUT_PATTERN.search(result.stderr.decode())
106
raise ValueError("Signing key fingerprint not found")
107
return m.group("fingerprint")
110
def find_changes_bugs(changes_url):
111
with urllib.request.urlopen(changes_url) as changes_fobj:
112
changes = debian.deb822.Changes(changes_fobj)
114
bugs_str = changes["Launchpad-Bugs-Fixed"]
117
return bugs_str.split()
120
def ensure_subscribed(person, bug, dry_run):
121
for subscription in bug.subscriptions:
123
subscription.person_link == person.self_link
124
and subscription.bug_notification_level == "Discussion"
126
logging.debug(f"{person.name} is already subscribed to {bug.id}")
128
logging.debug(f"Subscribing {person.name} to {bug.id}")
129
if "bot-stop-nagging" in bug.tags:
130
logging.debug("bot-stop-nagging detected; not subscribing")
133
bug.subscribe(level="Discussion", person=person)
136
def parse_fingerprints(output):
137
for line in output.decode().splitlines():
138
fields = line.split(":")
139
if fields[0] == "fpr":
143
def fetch_subkey_fingerprints(primary_fingerprints):
144
with tempfile.TemporaryDirectory() as tmpdir:
145
for primary_fingerprint in primary_fingerprints:
146
logging.debug(f"Fetching key for {primary_fingerprint}")
147
result = subprocess.run(
151
"keyserver.ubuntu.com",
155
env={"GNUPGHOME": tmpdir},
159
result.returncode == 2
160
and result.stderr == "gpg: keyserver receive failed: No data\n".encode()
162
# Some keys cannot be fetched. See: https://irclogs.ubuntu.com/2022/09/07/%23launchpad.html#t14:03
164
result.check_returncode()
165
result = subprocess.run(
170
"--with-fingerprint",
171
"--with-fingerprint", # duplicate to also give subkey fingerprints
174
env={"GNUPGHOME": tmpdir},
178
yield primary_fingerprint, (
180
for fpr in parse_fingerprints(result.stdout)
181
if fpr != primary_fingerprint
185
def determine_fingerprint_to_person_map(lp):
186
logging.debug("Fetching uploader key fingerprints")
187
fingerprint_to_person = {
188
gpg_key.fingerprint: person
189
for person, gpg_key in itertools.chain.from_iterable(
190
((person, gpg_key) for gpg_key in person.gpg_keys)
191
for person in lp.people["ubuntu-uploaders"].participants
194
logging.debug(f"{len(fingerprint_to_person)} fingerprints fetched")
196
# Add subkey fingerprints
197
logging.debug(f"Retrieving subkey fingerprints from keyserver.ubuntu.com")
198
fingerprint_to_person.update(
200
subkey_fingerprint: fingerprint_to_person[primary_fingerprint]
201
for primary_fingerprint, subkey_fingerprint in itertools.chain.from_iterable(
203
(primary_fingerprint, subkey_fingerprint)
204
for subkey_fingerprint in subkey_fingerprints
206
for primary_fingerprint, subkey_fingerprints in fetch_subkey_fingerprints(
207
fingerprint_to_person.keys()
212
logging.debug("Subkey fingerprints fetched")
214
# Some signers are only embedding 16 nibble keyids, so identify using these
215
# as well. It's easier to arrange collisions, but I think we can trust
216
# Ubuntu uploaders not to do this; and in the worst case scenario we'll
217
# only misidentify the uploader and give the wrong person the subscription
219
fingerprint_to_person.update(
221
fingerprint[-16:]: person
222
for fingerprint, person in fingerprint_to_person.items()
225
return fingerprint_to_person
228
if __name__ == "__main__":
229
parser = argparse.ArgumentParser()
230
parser.add_argument("--dry-run", action="store_true")
231
parser.add_argument("--debug", action="store_true")
232
args = parser.parse_args()
234
logging.basicConfig(level=logging.DEBUG)
236
lp = launchpadlib.launchpad.Launchpad.login_with(
237
"~racb sru_autosubscribe", "production", version="devel"
240
fingerprint_to_person = determine_fingerprint_to_person_map(lp)
244
for series in lp.distributions["ubuntu"].series
245
if series.status in {"Current Stable Release", "Supported"}
247
uploads = itertools.chain.from_iterable(
248
distro_series.getPackageUploads(pocket="Proposed", status="Unapproved")
249
for distro_series in distro_seriess
251
for upload in uploads:
252
logging.debug(f"Considering {upload}")
254
urls = get_upload_source_urls(upload)
256
logging.debug(f"Could not get source URLs for {upload}")
259
dsc_url = find_specific_url(urls, ".dsc")
261
logging.debug(f"Could not find .dsc for {upload}")
263
if not upload.changes_file_url:
264
logging.debug(f"Could not find changes file for {upload}")
266
bug_numbers = find_changes_bugs(upload.changes_file_url)
267
fingerprint = find_dsc_signing_fingerprint(dsc_url)
269
signer = fingerprint_to_person[fingerprint]
271
logging.debug(f"Could not find signer with fingerprint {fingerprint}")
273
for bug_number in bug_numbers:
275
bug = lp.bugs[bug_number]
277
logging.debug(f"Could not find bug {bug_number}")
279
ensure_subscribed(person=signer, bug=bug, dry_run=args.dry_run)