Coverage for dak/import_keyring.py: 69%
135 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-03 16:46 +0000
1"""Imports a keyring into the database"""
3# Copyright (C) 2007 Anthony Towns <aj@erisian.com.au>
4# Copyright (C) 2009 Mark Hymers <mhy@debian.org>
6# This program is free software; you can redistribute it and/or modify
7# it under the terms of the GNU General Public License as published by
8# the Free Software Foundation; either version 2 of the License, or
9# (at your option) any later version.
11# This program is distributed in the hope that it will be useful,
12# but WITHOUT ANY WARRANTY; without even the implied warranty of
13# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14# GNU General Public License for more details.
16# You should have received a copy of the GNU General Public License
17# along with this program; if not, write to the Free Software
18# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
20################################################################################
22import sys
24import apt_pkg
25from sqlalchemy import select, sql
27from daklib.config import Config
28from daklib.dbconn import DBConn, Fingerprint, Keyring, Uid, get_keyring
30# Globals
31Options = None
33################################################################################
36def get_uid_info(session):
37 byname = {}
38 byid = {}
39 q = session.execute(sql.text("SELECT id, uid, name FROM uid"))
40 for keyid, uid, name in q.fetchall():
41 byname[uid] = (keyid, name)
42 byid[keyid] = (uid, name)
44 return (byname, byid)
47def get_fingerprint_info(session):
48 fins = {}
49 q = session.execute(
50 sql.text("SELECT f.fingerprint, f.id, f.uid, f.keyring FROM fingerprint f")
51 )
52 for fingerprint, fingerprint_id, uid, keyring in q.fetchall():
53 fins[fingerprint] = (uid, fingerprint_id, keyring)
54 return fins
57def list_uids(session, pattern):
58 sql_pattern = f"%{pattern}%"
59 message = "List UIDs matching pattern %s" % sql_pattern
60 message += "\n" + ("=" * len(message))
61 print(message)
62 uid_query = select(Uid).where(Uid.uid.ilike(sql_pattern))
63 for uid in session.scalars(uid_query):
64 print("\nuid %s" % uid.uid)
65 for fp in uid.fingerprint:
66 print(" fingerprint %s" % fp.fingerprint)
67 keyring = "unknown"
68 if fp.keyring:
69 keyring = fp.keyring.keyring_name
70 print(" keyring %s" % keyring)
73################################################################################
76def usage(exit_code=0):
77 print(
78 """Usage: dak import-keyring [OPTION]... [KEYRING]
79 -h, --help show this help and exit.
80 -L, --import-ldap-users generate uid entries for keyring from LDAP
81 -U, --generate-users FMT generate uid entries from keyring as FMT
82 -l, --list-uids STRING list all uids matching *STRING*
83 -n, --no-action don't change database"""
84 )
85 sys.exit(exit_code)
88################################################################################
91def main():
92 global Options
94 cnf = Config()
95 Arguments = [
96 ("h", "help", "Import-Keyring::Options::Help"),
97 ("L", "import-ldap-users", "Import-Keyring::Options::Import-Ldap-Users"),
98 ("U", "generate-users", "Import-Keyring::Options::Generate-Users", "HasArg"),
99 ("l", "list-uids", "Import-Keyring::Options::List-UIDs", "HasArg"),
100 ("n", "no-action", "Import-Keyring::Options::No-Action"),
101 ]
103 for i in [
104 "help",
105 "report-changes",
106 "generate-users",
107 "import-ldap-users",
108 "list-uids",
109 "no-action",
110 ]:
111 key = "Import-Keyring::Options::%s" % i
112 if key not in cnf: 112 ↛ 103line 112 didn't jump to line 103
113 cnf[key] = ""
115 keyring_names = apt_pkg.parse_commandline(cnf.Cnf, Arguments, sys.argv) # type: ignore[attr-defined]
117 ### Parse options
119 Options = cnf.subtree("Import-Keyring::Options")
121 if Options["Help"]:
122 usage()
124 ### Initialise
125 session = DBConn().session()
127 if Options["List-UIDs"]: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 list_uids(session, Options["List-UIDs"])
129 sys.exit(0)
131 if len(keyring_names) != 1: 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true
132 usage(1)
134 ### Keep track of changes made
135 changes = [] # (uid, changes strings)
137 ### Cache all the existing fingerprint entries
138 db_fin_info = get_fingerprint_info(session)
140 ### Parse the keyring
142 keyringname = keyring_names[0]
143 keyring = get_keyring(keyringname, session)
144 if not keyring: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true
145 print("E: Can't load keyring %s from database" % keyringname)
146 sys.exit(1)
148 keyring.load_keys(keyringname)
150 ### Generate new uid entries if they're needed (from LDAP or the keyring)
151 if Options["Generate-Users"]: 151 ↛ 155line 151 didn't jump to line 155 because the condition on line 151 was always true
152 _, desuid_byid = keyring.generate_users_from_keyring(
153 Options["Generate-Users"], session
154 )
155 elif Options["Import-Ldap-Users"]:
156 _, desuid_byid = keyring.import_users_from_ldap(session)
157 else:
158 desuid_byid = {}
160 ### Cache all the existing uid entries
161 (db_uid_byname, db_uid_byid) = get_uid_info(session)
163 ### Update full names of applicable users
164 for keyid in desuid_byid.keys():
165 uid = (keyid, desuid_byid[keyid][0])
166 name = desuid_byid[keyid][1]
167 oname = db_uid_byid[keyid][1]
168 if name and oname != name:
169 changes.append((uid[1], "Full name: %s" % (name)))
170 session.execute(
171 sql.text("UPDATE uid SET name = :name WHERE id = :keyid"),
172 {"name": name, "keyid": keyid},
173 )
175 # The fingerprint table (fpr) points to a uid and a keyring.
176 # If the uid is being decided here (ldap/generate) we set it to it.
177 # Otherwise, if the fingerprint table already has a uid (which we've
178 # cached earlier), we preserve it.
179 # Otherwise we leave it as None
181 fpr = {}
182 for z in keyring.keys.keys():
183 keyid = db_uid_byname.get(keyring.keys[z].get("uid", None), [None])[0]
184 if keyid is None: 184 ↛ 185line 184 didn't jump to line 185 because the condition on line 184 was never true
185 keyid = db_fin_info.get(keyring.keys[z]["fingerprints"][0], [None])[0]
186 for y in keyring.keys[z]["fingerprints"]:
187 fpr[y] = (keyid, keyring.keyring_id)
189 # For any keys that used to be in this keyring, disassociate them.
190 # We don't change the uid, leaving that for historical info; if
191 # the id should change, it'll be set when importing another keyring.
193 for f, (u, fid, kr) in db_fin_info.items():
194 if kr != keyring.keyring_id: 194 ↛ 197line 194 didn't jump to line 197 because the condition on line 194 was always true
195 continue
197 if f in fpr:
198 continue
200 changes.append((db_uid_byid.get(u, [None])[0], "Removed key: %s" % (f)))
201 session.execute(
202 sql.text(
203 """UPDATE fingerprint
204 SET keyring = NULL
205 WHERE id = :fprid"""
206 ),
207 {"fprid": fid},
208 )
210 # For the keys in this keyring, add/update any fingerprints that've
211 # changed.
213 for f in fpr:
214 newuid = fpr[f][0]
215 newuiduid = db_uid_byid.get(newuid, [None])[0]
217 (olduid, oldfid, oldkid) = db_fin_info.get(f, [-1, -1, -1])
219 if olduid is None: 219 ↛ 220line 219 didn't jump to line 220 because the condition on line 219 was never true
220 olduid = -1
222 if oldkid is None: 222 ↛ 223line 222 didn't jump to line 223 because the condition on line 222 was never true
223 oldkid = -1
225 if oldfid == -1:
226 changes.append((newuiduid, "Added key: %s" % (f)))
227 fp = Fingerprint()
228 fp.fingerprint = f
229 fp.keyring_id = keyring.keyring_id
230 if newuid: 230 ↛ 233line 230 didn't jump to line 233 because the condition on line 230 was always true
231 fp.uid_id = newuid
233 session.add(fp)
234 session.flush()
236 else:
237 if newuid and olduid != newuid and olduid == -1: 237 ↛ 238line 237 didn't jump to line 238 because the condition on line 237 was never true
238 changes.append((newuiduid, "Linked key: %s" % f))
239 changes.append((newuiduid, " (formerly unowned)"))
240 session.execute(
241 sql.text("UPDATE fingerprint SET uid = :uid WHERE id = :fpr"),
242 {"uid": newuid, "fpr": oldfid},
243 )
245 # Don't move a key from a keyring with a higher priority to a lower one
246 if oldkid != keyring.keyring_id: 246 ↛ 213line 246 didn't jump to line 213 because the condition on line 246 was always true
247 movekey = False
248 if oldkid == -1: 248 ↛ 249line 248 didn't jump to line 249 because the condition on line 248 was never true
249 movekey = True
250 else:
251 oldkeyring = session.get(Keyring, oldkid)
252 if oldkeyring is None: 252 ↛ 253line 252 didn't jump to line 253 because the condition on line 252 was never true
253 print("ERROR: Cannot find old keyring with id %s" % oldkid)
254 sys.exit(1)
256 if oldkeyring.priority < keyring.priority: 256 ↛ 257line 256 didn't jump to line 257 because the condition on line 256 was never true
257 movekey = True
259 # Only change the keyring if it won't result in a loss of permissions
260 if movekey: 260 ↛ 261line 260 didn't jump to line 261 because the condition on line 260 was never true
261 session.execute(
262 sql.text(
263 """UPDATE fingerprint
264 SET keyring = :keyring
265 WHERE id = :fpr"""
266 ),
267 {"keyring": keyring.keyring_id, "fpr": oldfid},
268 )
270 session.flush()
272 else:
273 # movekey is only False if the old keyring was looked up
274 assert oldkeyring is not None
275 print(
276 "Key %s exists in both %s and %s keyrings. Not demoting."
277 % (f, oldkeyring.keyring_name, keyring.keyring_name)
278 )
280 # All done!
281 if Options["No-Action"]: 281 ↛ 282line 281 didn't jump to line 282 because the condition on line 281 was never true
282 session.rollback()
283 else:
284 session.commit()
286 # Print a summary
287 changesd = {}
288 for k, v in changes:
289 if k not in changesd:
290 changesd[k] = ""
291 changesd[k] += " %s\n" % (v)
293 for k in sorted(changesd):
294 print("%s\n%s\n" % (k, changesd[k]))