Coverage for dak/acl.py: 52%

1# Copyright (C) 2012, Ansgar Burchardt <ansgar@debian.org> 

2# Copyright (C) 2023 Emilio Pozuelo Monfort <pochu@debian.org> 

3# 

4# This program is free software; you can redistribute it and/or modify 

5# it under the terms of the GNU General Public License as published by 

6# the Free Software Foundation; either version 2 of the License, or 

7# (at your option) any later version. 

8# 

9# This program is distributed in the hope that it will be useful, 

10# but WITHOUT ANY WARRANTY; without even the implied warranty of 

11# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

12# GNU General Public License for more details. 

13# 

14# You should have received a copy of the GNU General Public License along 

15# with this program; if not, write to the Free Software Foundation, Inc., 

16# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. 

17 

18import os 

19import sys 

20from collections.abc import Iterable, Sequence 

21from typing import NoReturn, cast 

22 

23from sqlalchemy import CursorResult, delete, insert, select, sql 

24 

25from daklib.dbconn import ( 

26 ACL, 

27 ACLPerSource, 

28 ACLPerSuite, 

29 DBConn, 

30 Fingerprint, 

31 Keyring, 

32 Suite, 

33 Uid, 

34) 

35 

36 

37def usage(status: int = 0) -> NoReturn: 

38 print( 

39 """Usage: 

40 dak acl set-fingerprints <acl-name> 

41 dak acl export-per-source <acl-name> 

42 dak acl allow <acl-name> <fingerprint> <source>... 

43 dak acl deny <acl-name> <fingerprint> <source>... 

44 dak acl export-per-suite <acl-name> 

45 dak acl allow-suite <acl-name> <fingerprint> <suite>... 

46 dak acl deny-suite <acl-name> <fingerprint> <suite>... 

47 

48 set-fingerprints: 

49 Reads list of fingerprints from stdin and sets the ACL <acl-name> to these. 

50 Accepted input formats are "uid:<uid>", "name:<name>" and 

51 "fpr:<fingerprint>". 

52 

53 export-per-source: 

54 Export per source upload rights for ACL <acl-name>. 

55 

56 allow, deny: 

57 Grant (revoke) per-source upload rights for ACL <acl-name>. 

58 

59 export-per-suite: 

60 Export per suite upload rights for ACL <acl-name>. 

61 

62 allow-suite, deny-suite: 

63 Grant (revoke) per-suite upload rights for ACL <acl-name>. 

64""" 

65 ) 

66 sys.exit(status) 

67 

68 

69def get_fingerprint(entry: str, session) -> Sequence[Fingerprint]: 

70 """get fingerprint for given ACL entry 

71 

72 The entry is a string in one of these formats:: 

73 

74 uid:<uid> 

75 name:<name> 

76 fpr:<fingerprint> 

77 keyring:<keyring-name> 

78 

79 :param entry: ACL entry 

80 :param session: database session 

81 :return: fingerprint for the entry 

82 """ 

83 field, value = entry.split(":", 1) 

84 q = select(Fingerprint).join(Fingerprint.keyring).where(Keyring.active.is_(True)) 

85 

86 if field == "uid": 

87 q = q.join(Fingerprint.uid).where(Uid.uid == value) 

88 elif field == "name": 

89 q = q.join(Fingerprint.uid).where(Uid.name == value) 

90 elif field == "fpr": 

91 q = q.where(Fingerprint.fingerprint == value) 

92 elif field == "keyring": 

93 q = q.where(Keyring.keyring_name == value) 

94 else: 

95 raise Exception('Unknown selector "{0}".'.format(field)) 

96 

97 return list(session.scalars(q)) 

98 

99 

100def acl_set_fingerprints(acl_name: str, entries: Iterable[str]) -> None: 

101 session = DBConn().session() 

102 acl = session.execute(select(ACL).filter_by(name=acl_name)).scalar_one() 

103 

104 acl.fingerprints.clear() 

105 for entry in entries: 

106 entry = entry.strip() 

107 if entry.startswith("#") or len(entry) == 0: 

108 continue 

109 

110 fps = get_fingerprint(entry, session) 

111 if len(fps) == 0: 

112 print("Unknown key for '{0}'".format(entry)) 

113 else: 

114 acl.fingerprints.update(fps) 

115 

116 session.commit() 

117 

118 

119def acl_export_per_source(acl_name: str) -> None: 

120 session = DBConn().session() 

121 acl = session.execute(select(ACL).filter_by(name=acl_name)).scalar_one() 

122 

123 query = r""" 

124 SELECT 

125 f.fingerprint, 

126 (SELECT COALESCE(u.name, '') || ' <' || u.uid || '>' 

127 FROM uid u 

128 JOIN fingerprint f2 ON u.id = f2.uid 

129 WHERE f2.id = f.id) AS name, 

130 STRING_AGG( 

131 a.source 

132 || COALESCE(' (' || (SELECT fingerprint FROM fingerprint WHERE id = a.created_by_id) || ')', ''), 

133 E',\n ' ORDER BY a.source) 

134 FROM acl_per_source a 

135 JOIN fingerprint f ON a.fingerprint_id = f.id 

136 LEFT JOIN uid u ON f.uid = u.id 

137 WHERE a.acl_id = :acl_id 

138 GROUP BY f.id, f.fingerprint 

139 ORDER BY name 

140 """ 

141 

142 for row in session.execute(sql.text(query), {"acl_id": acl.id}): 

143 print("Fingerprint:", row[0]) 

144 print("Uid:", row[1]) 

145 print("Allow:", row[2]) 

146 print() 

147 

148 session.rollback() 

149 session.close() 

150 

151 

152def acl_export_per_suite(acl_name: str) -> None: 

153 session = DBConn().session() 

154 acl = session.execute(select(ACL).filter_by(name=acl_name)).scalar_one() 

155 

156 query = r""" 

157 SELECT 

158 f.fingerprint, 

159 (SELECT COALESCE(u.name, '') || ' <' || u.uid || '>' 

160 FROM uid u 

161 JOIN fingerprint f2 ON u.id = f2.uid 

162 WHERE f2.id = f.id) AS name, 

163 s.suite_name 

164 FROM acl_per_suite a 

165 JOIN fingerprint f ON a.fingerprint_id = f.id 

166 JOIN suite s ON a.suite_id = s.id 

167 LEFT JOIN uid u ON f.uid = u.id 

168 WHERE a.acl_id = :acl_id 

169 GROUP BY f.id, f.fingerprint, s.suite_name 

170 ORDER BY name 

171 """ 

172 

173 for row in session.execute(sql.text(query), {"acl_id": acl.id}): 

174 print("Fingerprint:", row[0]) 

175 print("Uid:", row[1]) 

176 print("Allow:", row[2]) 

177 print() 

178 

179 session.rollback() 

180 session.close() 

181 

182 

183def acl_allow(acl_name: str, fingerprint: str, sources: Iterable[str]) -> None: 

184 session = DBConn().session() 

185 

186 acl_id = session.execute(select(ACL.id).filter_by(name=acl_name)).scalar_one() 

187 fingerprint_id = session.execute( 

188 select(Fingerprint.fingerprint_id).filter_by(fingerprint=fingerprint) 

189 ).scalar_one() 

190 

191 # TODO: check that fpr is in ACL 

192 

193 data = [ 

194 { 

195 "acl_id": acl_id, 

196 "fingerprint_id": fingerprint_id, 

197 "source": source, 

198 "reason": "set by {} via CLI".format(os.environ.get("USER", "(unknown)")), 

199 } 

200 for source in sources 

201 ] 

202 

203 session.execute(insert(ACLPerSource), data) 

204 

205 session.commit() 

206 

207 

208def acl_allow_suite(acl_name: str, fingerprint: str, suites: Iterable[str]) -> None: 

209 session = DBConn().session() 

210 

211 acl_id = session.execute(select(ACL.id).filter_by(name=acl_name)).scalar_one() 

212 fingerprint_id = session.execute( 

213 select(Fingerprint.fingerprint_id).filter_by(fingerprint=fingerprint) 

214 ).scalar_one() 

215 

216 # TODO: check that fpr is in ACL 

217 

218 data = [] 

219 

220 for suite in suites: 

221 try: 

222 suite_id = session.execute( 

223 select(Suite.suite_id).filter_by(suite_name=suite) 

224 ).scalar_one() 

225 except: 

226 suite_id = session.execute( 

227 select(Suite.suite_id).filter_by(codename=suite) 

228 ).scalar_one() 

229 

230 data.append( 

231 { 

232 "acl_id": acl_id, 

233 "fingerprint_id": fingerprint_id, 

234 "suite_id": suite_id, 

235 "reason": "set by {} via CLI".format( 

236 os.environ.get("USER", "(unknown)") 

237 ), 

238 } 

239 ) 

240 

241 session.execute(insert(ACLPerSuite), data) 

242 

243 session.commit() 

244 

245 

246def acl_deny(acl_name: str, fingerprint: str, sources: Iterable[str]) -> None: 

247 session = DBConn().session() 

248 

249 acl_id = session.execute(select(ACL.id).filter_by(name=acl_name)).scalar_one() 

250 fingerprint_id = session.execute( 

251 select(Fingerprint.fingerprint_id).filter_by(fingerprint=fingerprint) 

252 ).scalar_one() 

253 

254 # TODO: check that fpr is in ACL 

255 

256 for source in sources: 

257 result = cast( 

258 CursorResult, 

259 session.execute( 

260 delete(ACLPerSource) 

261 .where(ACLPerSource.acl_id == acl_id) 

262 .where(ACLPerSource.fingerprint_id == fingerprint_id) 

263 .where(ACLPerSource.source == source) 

264 ), 

265 ) 

266 if result.rowcount < 1: 

267 print( 

268 "W: Tried to deny uploads of '{}', but was not allowed before.".format( 

269 source 

270 ) 

271 ) 

272 

273 session.commit() 

274 

275 

276def acl_deny_suite(acl_name: str, fingerprint: str, suites: Iterable[str]) -> None: 

277 session = DBConn().session() 

278 

279 acl_id = session.execute(select(ACL.id).filter_by(name=acl_name)).scalar_one() 

280 fingerprint_id = session.execute( 

281 select(Fingerprint.fingerprint_id).filter_by(fingerprint=fingerprint) 

282 ).scalar_one() 

283 

284 # TODO: check that fpr is in ACL 

285 

286 for suite in suites: 

287 try: 

288 suite_id = session.execute( 

289 select(Suite.suite_id).filter_by(suite_name=suite) 

290 ).scalar_one() 

291 except: 

292 suite_id = session.execute( 

293 select(Suite.suite_id).filter_by(codename=suite) 

294 ).scalar_one() 

295 

296 result = cast( 

297 CursorResult, 

298 session.execute( 

299 delete(ACLPerSuite) 

300 .where(ACLPerSuite.acl_id == acl_id) 

301 .where(ACLPerSuite.fingerprint_id == fingerprint_id) 

302 .where(ACLPerSuite.suite_id == suite_id) 

303 ), 

304 ) 

305 if result.rowcount < 1: 

306 print( 

307 "W: Tried to deny uploads for suite '{}', but was not allowed before.".format( 

308 suite 

309 ) 

310 ) 

311 

312 session.commit() 

313 

314 

315def main(argv=None): 

316 if argv is None: 316 ↛ 319line 316 didn't jump to line 319 because the condition on line 316 was always true

317 argv = sys.argv 

318 

319 if len(argv) > 1 and argv[1] in ("-h", "--help"): 

320 usage(0) 

321 

322 if len(argv) < 3: 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true

323 usage(1) 

324 

325 if argv[1] == "set-fingerprints": 325 ↛ 326line 325 didn't jump to line 326 because the condition on line 325 was never true

326 acl_set_fingerprints(argv[2], sys.stdin) 

327 elif argv[1] == "export-per-source": 

328 acl_export_per_source(argv[2]) 

329 elif argv[1] == "export-per-suite": 

330 acl_export_per_suite(argv[2]) 

331 elif argv[1] == "allow": 

332 acl_allow(argv[2], argv[3], argv[4:]) 

333 elif argv[1] == "deny": 333 ↛ 334line 333 didn't jump to line 334 because the condition on line 333 was never true

334 acl_deny(argv[2], argv[3], argv[4:]) 

335 elif argv[1] == "allow-suite": 335 ↛ 337line 335 didn't jump to line 337 because the condition on line 335 was always true

336 acl_allow_suite(argv[2], argv[3], argv[4:]) 

337 elif argv[1] == "deny-suite": 

338 acl_deny_suite(argv[2], argv[3], argv[4:]) 

339 else: 

340 usage(1)