Read the next I{count} records from the wire data and add them to the specified section. section_number: the section of the message to which to add records count: the number of records to read
(self, section_number, count)
| 1167 | self.errors.append(MessageError(e, self.parser.current)) |
| 1168 | |
| 1169 | def _get_section(self, section_number, count): |
| 1170 | """Read the next I{count} records from the wire data and add them to |
| 1171 | the specified section. |
| 1172 | |
| 1173 | section_number: the section of the message to which to add records |
| 1174 | count: the number of records to read |
| 1175 | """ |
| 1176 | assert self.message is not None |
| 1177 | section = self.message.sections[section_number] |
| 1178 | force_unique = self.one_rr_per_rrset |
| 1179 | for i in range(count): |
| 1180 | rr_start = self.parser.current |
| 1181 | absolute_name = self.parser.get_name() |
| 1182 | if self.message.origin is not None: |
| 1183 | name = absolute_name.relativize(self.message.origin) |
| 1184 | else: |
| 1185 | name = absolute_name |
| 1186 | (rdtype, rdclass, ttl, rdlen) = self.parser.get_struct("!HHIH") |
| 1187 | if rdtype in (dns.rdatatype.OPT, dns.rdatatype.TSIG): |
| 1188 | ( |
| 1189 | rdclass, |
| 1190 | rdtype, |
| 1191 | deleting, |
| 1192 | empty, |
| 1193 | ) = self.message._parse_special_rr_header( |
| 1194 | section_number, count, i, name, rdclass, rdtype |
| 1195 | ) |
| 1196 | else: |
| 1197 | (rdclass, rdtype, deleting, empty) = self.message._parse_rr_header( |
| 1198 | section_number, name, rdclass, rdtype |
| 1199 | ) |
| 1200 | rdata_start = self.parser.current |
| 1201 | try: |
| 1202 | if empty: |
| 1203 | if rdlen > 0: |
| 1204 | raise dns.exception.FormError |
| 1205 | rd = None |
| 1206 | covers = dns.rdatatype.NONE |
| 1207 | else: |
| 1208 | with self.parser.restrict_to(rdlen): |
| 1209 | rd = dns.rdata.from_wire_parser( |
| 1210 | rdclass, # pyright: ignore |
| 1211 | rdtype, |
| 1212 | self.parser, |
| 1213 | self.message.origin, |
| 1214 | ) |
| 1215 | covers = rd.covers() |
| 1216 | if self.message.xfr and rdtype == dns.rdatatype.SOA: |
| 1217 | force_unique = True |
| 1218 | if rdtype == dns.rdatatype.OPT: |
| 1219 | self.message.opt = dns.rrset.from_rdata(name, ttl, rd) |
| 1220 | elif rdtype == dns.rdatatype.TSIG: |
| 1221 | trd = cast(dns.rdtypes.ANY.TSIG.TSIG, rd) |
| 1222 | if self.keyring is None or self.keyring is True: |
| 1223 | raise UnknownTSIGKey("got signed message without keyring") |
| 1224 | elif isinstance(self.keyring, dict): |
| 1225 | key = self.keyring.get(absolute_name) |
| 1226 | if isinstance(key, bytes): |
no test coverage detected