Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dns/transaction.py: 27%

1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license 

2 

3import collections 

4from collections.abc import Callable, Iterator 

5from typing import Any 

6 

7import dns.exception 

8import dns.name 

9import dns.node 

10import dns.rdata 

11import dns.rdataclass 

12import dns.rdataset 

13import dns.rdatatype 

14import dns.rrset 

15import dns.serial 

16import dns.ttl 

17 

18 

19class TransactionManager: 

20 def reader(self) -> "Transaction": 

21 """Begin a read-only transaction.""" 

22 raise NotImplementedError # pragma: no cover 

23 

24 def writer(self, replacement: bool = False) -> "Transaction": 

25 """Begin a writable transaction. 

26 

27 :param bool replacement: If ``True``, the content of the transaction 

28 completely replaces any prior content. If ``False`` (the default), 

29 the content of the transaction updates the existing content. 

30 """ 

31 raise NotImplementedError # pragma: no cover 

32 

33 def origin_information( 

34 self, 

35 ) -> tuple[dns.name.Name | None, bool, dns.name.Name | None]: 

36 """Return origin information for the transaction manager. 

37 

38 This information is used by code working with transactions to 

39 allow it to coordinate relativization. The transaction code 

40 itself takes what it gets (i.e. does not change name 

41 relativity). 

42 

43 :returns: A tuple ``(absolute_origin, relativize, effective_origin)`` 

44 giving the absolute name of the default origin for any relative 

45 domain names, whether names should be relativized, and the 

46 "effective origin". The effective origin is the absolute origin if 

47 *relativize* is ``False``, and the empty name if *relativize* is 

48 ``True``. (The effective origin is provided even though it can be 

49 computed from the absolute origin and relativize setting because it 

50 avoids a lot of code duplication.) If the returned names are 

51 ``None``, then no origin information is available. 

52 :rtype: tuple[:py:class:`dns.name.Name` or ``None``, bool, 

53 :py:class:`dns.name.Name` or ``None``] 

54 """ 

55 raise NotImplementedError # pragma: no cover 

56 

57 def get_class(self) -> dns.rdataclass.RdataClass: 

58 """The class of the transaction manager.""" 

59 raise NotImplementedError # pragma: no cover 

60 

61 def from_wire_origin(self) -> dns.name.Name | None: 

62 """Origin to use in from_wire() calls.""" 

63 absolute_origin, relativize, _ = self.origin_information() 

64 if relativize: 

65 return absolute_origin 

66 else: 

67 return None 

68 

69 

70class DeleteNotExact(dns.exception.DNSException): 

71 """Existing data did not match data specified by an exact delete.""" 

72 

73 

74class ReadOnly(dns.exception.DNSException): 

75 """Tried to write to a read-only transaction.""" 

76 

77 

78class AlreadyEnded(dns.exception.DNSException): 

79 """Tried to use an already-ended transaction.""" 

80 

81 

82def _ensure_immutable_rdataset(rdataset) -> dns.rdataset.ImmutableRdataset | None: 

83 if rdataset is None or isinstance(rdataset, dns.rdataset.ImmutableRdataset): 

84 return rdataset 

85 return dns.rdataset.ImmutableRdataset(rdataset) 

86 

87 

88def _ensure_immutable_node(node) -> dns.node.ImmutableNode | None: 

89 if node is None or node.is_immutable(): 

90 return node 

91 return dns.node.ImmutableNode(node) 

92 

93 

94CheckPutRdatasetType = Callable[ 

95 ["Transaction", dns.name.Name, dns.rdataset.Rdataset], None 

96] 

97CheckDeleteRdatasetType = Callable[ 

98 ["Transaction", dns.name.Name, dns.rdatatype.RdataType, dns.rdatatype.RdataType], 

99 None, 

100] 

101CheckDeleteNameType = Callable[["Transaction", dns.name.Name], None] 

102 

103 

104class Transaction: 

105 def __init__( 

106 self, 

107 manager: TransactionManager, 

108 replacement: bool = False, 

109 read_only: bool = False, 

110 ): 

111 self.manager = manager 

112 self.replacement = replacement 

113 self.read_only = read_only 

114 self.unicode: set[str] = set() 

115 self._ended = False 

116 self._check_put_rdataset: list[CheckPutRdatasetType] = [] 

117 self._check_delete_rdataset: list[CheckDeleteRdatasetType] = [] 

118 self._check_delete_name: list[CheckDeleteNameType] = [] 

119 

120 # 

121 # This is the high level API 

122 # 

123 # Note that we currently use non-immutable types in the return type signature to 

124 # avoid covariance problems, e.g. if the caller has a list[Rdataset], mypy will be 

125 # unhappy if we return an ImmutableRdataset. 

126 

127 def get( 

128 self, 

129 name: dns.name.Name | str | None, 

130 rdtype: dns.rdatatype.RdataType | str, 

131 covers: dns.rdatatype.RdataType | str = dns.rdatatype.NONE, 

132 ) -> dns.rdataset.Rdataset | None: 

133 """Return the rdataset associated with *name*, *rdtype*, and *covers*, 

134 or ``None`` if not found. 

135 

136 Note that the returned rdataset is immutable. 

137 """ 

138 self._check_ended() 

139 if isinstance(name, str): 

140 name = dns.name.from_text(name, None) 

141 rdtype = dns.rdatatype.RdataType.make(rdtype) 

142 covers = dns.rdatatype.RdataType.make(covers) 

143 rdataset = self._get_rdataset(name, rdtype, covers) 

144 return _ensure_immutable_rdataset(rdataset) 

145 

146 def get_node(self, name: dns.name.Name) -> dns.node.Node | None: 

147 """Return the node at *name*, if any. 

148 

149 :returns: An immutable node, or ``None`` if *name* does not exist. 

150 :rtype: :py:class:`dns.node.ImmutableNode` or ``None`` 

151 """ 

152 return _ensure_immutable_node(self._get_node(name)) 

153 

154 def _check_read_only(self) -> None: 

155 if self.read_only: 

156 raise ReadOnly 

157 

158 def add(self, *args: Any) -> None: 

159 """Add records. 

160 

161 The arguments may be: 

162 

163 - rrset 

164 

165 - name, rdataset... 

166 

167 - name, ttl, rdata... 

168 """ 

169 self._check_ended() 

170 self._check_read_only() 

171 self._add(False, args) 

172 

173 def replace(self, *args: Any) -> None: 

174 """Replace the existing rdataset at the name with the specified 

175 rdataset, or add the specified rdataset if there was no existing 

176 rdataset. 

177 

178 The arguments may be: 

179 

180 - rrset 

181 

182 - name, rdataset... 

183 

184 - name, ttl, rdata... 

185 

186 Note that if you want to replace the entire node, you should do 

187 a delete of the name followed by one or more calls to add() or 

188 replace(). 

189 """ 

190 self._check_ended() 

191 self._check_read_only() 

192 self._add(True, args) 

193 

194 def delete(self, *args: Any) -> None: 

195 """Delete records. 

196 

197 It is not an error if some of the records are not in the existing 

198 set. 

199 

200 The arguments may be: 

201 

202 - rrset 

203 

204 - name 

205 

206 - name, rdatatype, [covers] 

207 

208 - name, rdataset... 

209 

210 - name, rdata... 

211 """ 

212 self._check_ended() 

213 self._check_read_only() 

214 self._delete(False, args) 

215 

216 def delete_exact(self, *args: Any) -> None: 

217 """Delete records. 

218 

219 The arguments may be: 

220 

221 - rrset 

222 

223 - name 

224 

225 - name, rdatatype, [covers] 

226 

227 - name, rdataset... 

228 

229 - name, rdata... 

230 

231 :raises: :py:exc:`dns.transaction.DeleteNotExact` if some of the 

232 records are not in the existing set. 

233 """ 

234 self._check_ended() 

235 self._check_read_only() 

236 self._delete(True, args) 

237 

238 def name_exists(self, name: dns.name.Name | str) -> bool: 

239 """Does the specified name exist?""" 

240 self._check_ended() 

241 if isinstance(name, str): 

242 name = dns.name.from_text(name, None) 

243 return self._name_exists(name) 

244 

245 def update_serial( 

246 self, 

247 value: int = 1, 

248 relative: bool = True, 

249 name: dns.name.Name = dns.name.empty, 

250 ) -> None: 

251 """Update the serial number. 

252 

253 :param int value: An increment if *relative* is ``True``, or the 

254 actual value to set if *relative* is ``False``. 

255 :param bool relative: Whether *value* is a relative increment or an 

256 absolute value. The default is ``True``. 

257 :param name: The name of the SOA record to update. 

258 :type name: :py:class:`dns.name.Name` 

259 :raises KeyError: If there is no SOA rdataset at *name*. 

260 :raises ValueError: If *value* is negative or if the increment is 

261 so large that it would cause the new serial to be less than the 

262 prior value. 

263 """ 

264 self._check_ended() 

265 if value < 0: 

266 raise ValueError("negative update_serial() value") 

267 if isinstance(name, str): 

268 name = dns.name.from_text(name, None) 

269 rdataset = self._get_rdataset(name, dns.rdatatype.SOA, dns.rdatatype.NONE) 

270 if rdataset is None or len(rdataset) == 0: 

271 raise KeyError 

272 if relative: 

273 serial = dns.serial.Serial(rdataset[0].serial) + value 

274 else: 

275 serial = dns.serial.Serial(value) 

276 serial = serial.value # convert back to int 

277 if serial == 0: 

278 serial = 1 

279 rdata = rdataset[0].replace(serial=serial) 

280 new_rdataset = dns.rdataset.from_rdata(rdataset.ttl, rdata) 

281 self.replace(name, new_rdataset) 

282 

283 def add_unicode(self, attribute: str): 

284 self.unicode.add(attribute.upper()) 

285 

286 def __iter__(self): 

287 self._check_ended() 

288 return self._iterate_rdatasets() 

289 

290 def changed(self) -> bool: 

291 """Has this transaction changed anything? 

292 

293 For read-only transactions, the result is always ``False``. 

294 

295 For writable transactions, the result is ``True`` if at some time 

296 during the life of the transaction, the content was changed. 

297 """ 

298 self._check_ended() 

299 return self._changed() 

300 

301 def commit(self) -> None: 

302 """Commit the transaction. 

303 

304 Normally transactions are used as context managers and commit 

305 or rollback automatically, but it may be done explicitly if needed. 

306 

307 :raises: :py:exc:`dns.transaction.AlreadyEnded` if the transaction 

308 has already been committed or rolled back. 

309 :raises: An exception if the commit fails, in which case the 

310 transaction is also rolled back. 

311 """ 

312 self._end(True) 

313 

314 def rollback(self) -> None: 

315 """Rollback the transaction. 

316 

317 Normally transactions are used as context managers and commit 

318 or rollback automatically, but it may be done explicitly if needed. 

319 Rollback cannot otherwise fail. 

320 

321 :raises: :py:exc:`dns.transaction.AlreadyEnded` if the transaction 

322 has already been committed or rolled back. 

323 """ 

324 self._end(False) 

325 

326 def check_put_rdataset(self, check: CheckPutRdatasetType) -> None: 

327 """Call *check* before putting (storing) an rdataset. 

328 

329 The function is called with the transaction, the name, and the rdataset. 

330 

331 The check function may safely make non-mutating transaction method 

332 calls, but behavior is undefined if mutating transaction methods are 

333 called. The check function should raise an exception if it objects to 

334 the put, and otherwise should return ``None``. 

335 """ 

336 self._check_put_rdataset.append(check) 

337 

338 def check_delete_rdataset(self, check: CheckDeleteRdatasetType) -> None: 

339 """Call *check* before deleting an rdataset. 

340 

341 The function is called with the transaction, the name, the rdatatype, 

342 and the covered rdatatype. 

343 

344 The check function may safely make non-mutating transaction method 

345 calls, but behavior is undefined if mutating transaction methods are 

346 called. The check function should raise an exception if it objects to 

347 the put, and otherwise should return ``None``. 

348 """ 

349 self._check_delete_rdataset.append(check) 

350 

351 def check_delete_name(self, check: CheckDeleteNameType) -> None: 

352 """Call *check* before putting (storing) an rdataset. 

353 

354 The function is called with the transaction and the name. 

355 

356 The check function may safely make non-mutating transaction method 

357 calls, but behavior is undefined if mutating transaction methods are 

358 called. The check function should raise an exception if it objects to 

359 the put, and otherwise should return ``None``. 

360 """ 

361 self._check_delete_name.append(check) 

362 

363 def iterate_rdatasets( 

364 self, 

365 ) -> Iterator[tuple[dns.name.Name, dns.rdataset.Rdataset]]: 

366 """Iterate all the rdatasets in the transaction, returning 

367 (:py:class:`dns.name.Name`, :py:class:`dns.rdataset.Rdataset`) tuples. 

368 

369 Note that as is usual with Python iterators, adding or removing items 

370 while iterating will invalidate the iterator and may raise 

371 :py:exc:`RuntimeError` or fail to iterate over all entries.""" 

372 self._check_ended() 

373 return self._iterate_rdatasets() 

374 

375 def iterate_names(self) -> Iterator[dns.name.Name]: 

376 """Iterate all the names in the transaction. 

377 

378 Note that as is usual with Python iterators, adding or removing names 

379 while iterating will invalidate the iterator and may raise 

380 :py:exc:`RuntimeError` or fail to iterate over all entries.""" 

381 self._check_ended() 

382 return self._iterate_names() 

383 

384 # 

385 # Helper methods 

386 # 

387 

388 def _raise_if_not_empty(self, method, args): 

389 if len(args) != 0: 

390 raise TypeError(f"extra parameters to {method}") 

391 

392 def _rdataset_from_args(self, method, deleting, args): 

393 try: 

394 arg = args.popleft() 

395 if isinstance(arg, dns.rrset.RRset): 

396 rdataset = arg.to_rdataset() 

397 elif isinstance(arg, dns.rdataset.Rdataset): 

398 rdataset = arg 

399 else: 

400 if deleting: 

401 ttl = 0 

402 else: 

403 if isinstance(arg, int): 

404 ttl = arg 

405 if ttl > dns.ttl.MAX_TTL: 

406 raise ValueError(f"{method}: TTL value too big") 

407 else: 

408 raise TypeError(f"{method}: expected a TTL") 

409 arg = args.popleft() 

410 if isinstance(arg, dns.rdata.Rdata): 

411 rdataset = dns.rdataset.from_rdata(ttl, arg) 

412 else: 

413 raise TypeError(f"{method}: expected an Rdata") 

414 return rdataset 

415 except IndexError: 

416 if deleting: 

417 return None 

418 else: 

419 # reraise 

420 raise TypeError(f"{method}: expected more arguments") 

421 

422 def _add(self, replace, args): 

423 if replace: 

424 method = "replace()" 

425 else: 

426 method = "add()" 

427 try: 

428 args = collections.deque(args) 

429 arg = args.popleft() 

430 if isinstance(arg, str): 

431 arg = dns.name.from_text(arg, None) 

432 if isinstance(arg, dns.name.Name): 

433 name = arg 

434 rdataset = self._rdataset_from_args(method, False, args) 

435 elif isinstance(arg, dns.rrset.RRset): 

436 rrset = arg 

437 name = rrset.name 

438 # rrsets are also rdatasets, but they don't print the 

439 # same and can't be stored in nodes, so convert. 

440 rdataset = rrset.to_rdataset() 

441 else: 

442 raise TypeError( 

443 f"{method} requires a name or RRset as the first argument" 

444 ) 

445 assert rdataset is not None # for type checkers 

446 if rdataset.rdclass != self.manager.get_class(): 

447 raise ValueError(f"{method} has objects of wrong RdataClass") 

448 if rdataset.rdtype == dns.rdatatype.SOA: 

449 _, _, origin = self._origin_information() 

450 if name != origin: 

451 raise ValueError(f"{method} has non-origin SOA") 

452 self._raise_if_not_empty(method, args) 

453 if not replace: 

454 existing = self._get_rdataset(name, rdataset.rdtype, rdataset.covers) 

455 if existing is not None: 

456 if isinstance(existing, dns.rdataset.ImmutableRdataset): 

457 trds = dns.rdataset.Rdataset( 

458 existing.rdclass, existing.rdtype, existing.covers 

459 ) 

460 trds.update(existing) 

461 existing = trds 

462 rdataset = existing.union(rdataset) 

463 self._checked_put_rdataset(name, rdataset) 

464 except IndexError: 

465 raise TypeError(f"not enough parameters to {method}") 

466 

467 def _delete(self, exact, args): 

468 if exact: 

469 method = "delete_exact()" 

470 else: 

471 method = "delete()" 

472 try: 

473 args = collections.deque(args) 

474 arg = args.popleft() 

475 if isinstance(arg, str): 

476 arg = dns.name.from_text(arg, None) 

477 if isinstance(arg, dns.name.Name): 

478 name = arg 

479 if len(args) > 0 and ( 

480 isinstance(args[0], int) or isinstance(args[0], str) 

481 ): 

482 # deleting by type and (optionally) covers 

483 rdtype = dns.rdatatype.RdataType.make(args.popleft()) 

484 if len(args) > 0: 

485 covers = dns.rdatatype.RdataType.make(args.popleft()) 

486 else: 

487 covers = dns.rdatatype.NONE 

488 self._raise_if_not_empty(method, args) 

489 existing = self._get_rdataset(name, rdtype, covers) 

490 if existing is None: 

491 if exact: 

492 raise DeleteNotExact(f"{method}: missing rdataset") 

493 else: 

494 self._checked_delete_rdataset(name, rdtype, covers) 

495 return 

496 else: 

497 rdataset = self._rdataset_from_args(method, True, args) 

498 elif isinstance(arg, dns.rrset.RRset): 

499 rdataset = arg # rrsets are also rdatasets 

500 name = rdataset.name 

501 else: 

502 raise TypeError( 

503 f"{method} requires a name or RRset as the first argument" 

504 ) 

505 self._raise_if_not_empty(method, args) 

506 if rdataset: 

507 if rdataset.rdclass != self.manager.get_class(): 

508 raise ValueError(f"{method} has objects of wrong RdataClass") 

509 existing = self._get_rdataset(name, rdataset.rdtype, rdataset.covers) 

510 if existing is not None: 

511 if exact: 

512 intersection = existing.intersection(rdataset) 

513 if intersection != rdataset: 

514 raise DeleteNotExact(f"{method}: missing rdatas") 

515 rdataset = existing.difference(rdataset) 

516 if len(rdataset) == 0: 

517 self._checked_delete_rdataset( 

518 name, rdataset.rdtype, rdataset.covers 

519 ) 

520 else: 

521 self._checked_put_rdataset(name, rdataset) 

522 elif exact: 

523 raise DeleteNotExact(f"{method}: missing rdataset") 

524 else: 

525 if exact and not self._name_exists(name): 

526 raise DeleteNotExact(f"{method}: name not known") 

527 self._checked_delete_name(name) 

528 except IndexError: 

529 raise TypeError(f"not enough parameters to {method}") 

530 

531 def _check_ended(self): 

532 if self._ended: 

533 raise AlreadyEnded 

534 

535 def _end(self, commit): 

536 self._check_ended() 

537 try: 

538 self._end_transaction(commit) 

539 finally: 

540 self._ended = True 

541 

542 def _checked_put_rdataset(self, name, rdataset): 

543 for check in self._check_put_rdataset: 

544 check(self, name, rdataset) 

545 self._put_rdataset(name, rdataset) 

546 

547 def _checked_delete_rdataset(self, name, rdtype, covers): 

548 for check in self._check_delete_rdataset: 

549 check(self, name, rdtype, covers) 

550 self._delete_rdataset(name, rdtype, covers) 

551 

552 def _checked_delete_name(self, name): 

553 for check in self._check_delete_name: 

554 check(self, name) 

555 self._delete_name(name) 

556 

557 # 

558 # Transactions are context managers. 

559 # 

560 

561 def __enter__(self): 

562 return self 

563 

564 def __exit__(self, exc_type, exc_val, exc_tb): 

565 if not self._ended: 

566 if exc_type is None: 

567 self.commit() 

568 else: 

569 self.rollback() 

570 return False 

571 

572 # 

573 # This is the low level API, which must be implemented by subclasses 

574 # of Transaction. 

575 # 

576 

577 def _get_rdataset(self, name, rdtype, covers): 

578 """Return the rdataset associated with *name*, *rdtype*, and *covers*, 

579 or ``None`` if not found. 

580 """ 

581 raise NotImplementedError # pragma: no cover 

582 

583 def _put_rdataset(self, name, rdataset): 

584 """Store the rdataset.""" 

585 raise NotImplementedError # pragma: no cover 

586 

587 def _delete_name(self, name): 

588 """Delete all data associated with *name*. 

589 

590 It is not an error if the name does not exist. 

591 """ 

592 raise NotImplementedError # pragma: no cover 

593 

594 def _delete_rdataset(self, name, rdtype, covers): 

595 """Delete all data associated with *name*, *rdtype*, and *covers*. 

596 

597 It is not an error if the rdataset does not exist. 

598 """ 

599 raise NotImplementedError # pragma: no cover 

600 

601 def _name_exists(self, name): 

602 """Does name exist? 

603 

604 :rtype: bool 

605 """ 

606 raise NotImplementedError # pragma: no cover 

607 

608 def _changed(self): 

609 """Has this transaction changed anything?""" 

610 raise NotImplementedError # pragma: no cover 

611 

612 def _end_transaction(self, commit): 

613 """End the transaction. 

614 

615 :param bool commit: If ``True``, commit the transaction; otherwise 

616 roll it back. If committing and the commit fails, then roll back 

617 and raise an exception. 

618 """ 

619 raise NotImplementedError # pragma: no cover 

620 

621 def _set_origin(self, origin): 

622 """Set the origin. 

623 

624 This method is called when reading a possibly relativized 

625 source, and an origin setting operation occurs (e.g. $ORIGIN 

626 in a zone file). 

627 """ 

628 raise NotImplementedError # pragma: no cover 

629 

630 def _iterate_rdatasets(self): 

631 """Return an iterator that yields (name, rdataset) tuples.""" 

632 raise NotImplementedError # pragma: no cover 

633 

634 def _iterate_names(self): 

635 """Return an iterator that yields a name.""" 

636 raise NotImplementedError # pragma: no cover 

637 

638 def _get_node(self, name): 

639 """Return the node at *name*, if any. 

640 

641 :returns: The node, or ``None`` if *name* does not exist. 

642 :rtype: :py:class:`dns.node.Node` or ``None`` 

643 """ 

644 raise NotImplementedError # pragma: no cover 

645 

646 # 

647 # Low-level API with a default implementation, in case a subclass needs 

648 # to override. 

649 # 

650 

651 def _origin_information(self): 

652 # This is only used by _add() 

653 return self.manager.origin_information() 

654 

655 

656class TooManyChanges(dns.exception.DNSException): 

657 """Too many changes""" 

658 

659 

660class TransactionLimiter: 

661 """Transaction setup that enforces a maximum number of changes in the transaction. 

662 

663 Each rdataset put or deleted counts as a change. Each whole name deletion counts 

664 as a change. 

665 

666 The limiter is meant to ensure that reading a zonefile from an untrusted source 

667 cannot cause too many changes, e.g. via large $GENERATE statements. It is not 

668 meant to limit the total number of records in a zone or the memory used by the 

669 zone. 

670 """ 

671 

672 def __init__(self, limit: int): 

673 super().__init__() 

674 self.limit = limit 

675 self.changes: int = 0 

676 

677 def _check_limit(self): 

678 if self.changes >= self.limit: 

679 raise TooManyChanges(f"limit is {self.limit} changes") 

680 

681 def _check_put_rdataset( 

682 self, txn: Transaction, name: dns.name.Name, rdataset: dns.rdataset.Rdataset 

683 ): 

684 self.changes += 1 

685 self._check_limit() 

686 

687 def _check_delete_rdataset( 

688 self, 

689 txn: Transaction, 

690 name: dns.name.Name, 

691 type: dns.rdatatype.RdataType, 

692 covers: dns.rdatatype.RdataType | None, 

693 ): 

694 self.changes += 1 

695 self._check_limit() 

696 

697 def _check_delete_name( 

698 self, 

699 txn: Transaction, 

700 name: dns.name.Name, 

701 ): 

702 self.changes += 1 

703 self._check_limit() 

704 

705 def __call__(self, txn: Transaction): 

706 txn.check_put_rdataset(self._check_put_rdataset) 

707 txn.check_delete_rdataset(self._check_delete_rdataset) 

708 txn.check_delete_name(self._check_delete_name)