Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dns/tokenizer.py: 18%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
3# Copyright (C) 2003-2017 Nominum, Inc.
4#
5# Permission to use, copy, modify, and distribute this software and its
6# documentation for any purpose with or without fee is hereby granted,
7# provided that the above copyright notice and this permission notice
8# appear in all copies.
9#
10# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
11# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
12# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
13# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
14# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
15# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
16# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
18"""Tokenize DNS zone file format"""
20import io
21import sys
22from typing import Any
24import dns._text_util
25import dns.exception
26import dns.name
27import dns.ttl
29_DELIMITERS = {" ", "\t", "\n", ";", "(", ")", '"'}
30_QUOTING_DELIMITERS = {'"'}
32EOF = 0
33EOL = 1
34WHITESPACE = 2
35IDENTIFIER = 3
36QUOTED_STRING = 4
37COMMENT = 5
38DELIMITER = 6
41class UngetBufferFull(dns.exception.DNSException):
42 """An attempt was made to unget a token when the unget buffer was full."""
45class Token:
46 """A DNS zone file format token.
48 ttype: The token type
49 value: The token value
50 has_escape: Does the token value contain escapes?
51 """
53 def __init__(
54 self,
55 ttype: int,
56 value: Any = "",
57 has_escape: bool = False,
58 comment: str | None = None,
59 ):
60 """Initialize a token instance."""
62 self.ttype = ttype
63 self.value = value
64 self.has_escape = has_escape
65 self.comment = comment
67 def is_eof(self) -> bool:
68 return self.ttype == EOF
70 def is_eol(self) -> bool:
71 return self.ttype == EOL
73 def is_whitespace(self) -> bool:
74 return self.ttype == WHITESPACE
76 def is_identifier(self) -> bool:
77 return self.ttype == IDENTIFIER
79 def is_quoted_string(self) -> bool:
80 return self.ttype == QUOTED_STRING
82 def is_comment(self) -> bool:
83 return self.ttype == COMMENT
85 def is_delimiter(self) -> bool: # pragma: no cover (we don't return delimiters yet)
86 return self.ttype == DELIMITER
88 def is_eol_or_eof(self) -> bool:
89 return self.ttype == EOL or self.ttype == EOF
91 def __eq__(self, other):
92 if not isinstance(other, Token):
93 return False
94 return self.ttype == other.ttype and self.value == other.value
96 def __ne__(self, other):
97 if not isinstance(other, Token):
98 return True
99 return self.ttype != other.ttype or self.value != other.value
101 def __str__(self):
102 return f'{self.ttype} "{self.value}"'
104 def unescape(self) -> "Token":
105 if not self.has_escape:
106 return self
107 unescaped = ""
108 l = len(self.value)
109 i = 0
110 while i < l:
111 c = self.value[i]
112 i += 1
113 if c == "\\":
114 if i >= l: # pragma: no cover (can't happen via get())
115 raise dns.exception.UnexpectedEnd
116 c = self.value[i]
117 i += 1
118 if dns._text_util.is_ascii_digit(c):
119 if i >= l:
120 raise dns.exception.UnexpectedEnd
121 c2 = self.value[i]
122 i += 1
123 if i >= l:
124 raise dns.exception.UnexpectedEnd
125 c3 = self.value[i]
126 i += 1
127 if not (
128 dns._text_util.is_ascii_digit(c2)
129 and dns._text_util.is_ascii_digit(c3)
130 ):
131 raise dns.exception.SyntaxError
132 codepoint = int(c) * 100 + int(c2) * 10 + int(c3)
133 if codepoint > 255:
134 raise dns.exception.SyntaxError
135 c = chr(codepoint)
136 unescaped += c
137 return Token(self.ttype, unescaped)
139 def unescape_to_bytes(self) -> "Token":
140 # We used to use unescape() for TXT-like records, but this
141 # caused problems as we'd process DNS escapes into Unicode code
142 # points instead of byte values, and then a to_text() of the
143 # processed data would not equal the original input. For
144 # example, \226 in the TXT record would have a to_text() of
145 # \195\162 because we applied UTF-8 encoding to Unicode code
146 # point 226.
147 #
148 # We now apply escapes while converting directly to bytes,
149 # avoiding this double encoding.
150 #
151 # This code also handles cases where the unicode input has
152 # non-ASCII code-points in it by converting it to UTF-8. TXT
153 # records aren't defined for Unicode, but this is the best we
154 # can do to preserve meaning. For example,
155 #
156 # foo\u200bbar
157 #
158 # (where \u200b is Unicode code point 0x200b) will be treated
159 # as if the input had been the UTF-8 encoding of that string,
160 # namely:
161 #
162 # foo\226\128\139bar
163 #
164 unescaped = b""
165 l = len(self.value)
166 i = 0
167 while i < l:
168 c = self.value[i]
169 i += 1
170 if c == "\\":
171 if i >= l: # pragma: no cover (can't happen via get())
172 raise dns.exception.UnexpectedEnd
173 c = self.value[i]
174 i += 1
175 if dns._text_util.is_ascii_digit(c):
176 if i >= l:
177 raise dns.exception.UnexpectedEnd
178 c2 = self.value[i]
179 i += 1
180 if i >= l:
181 raise dns.exception.UnexpectedEnd
182 c3 = self.value[i]
183 i += 1
184 if not (
185 dns._text_util.is_ascii_digit(c2)
186 and dns._text_util.is_ascii_digit(c3)
187 ):
188 raise dns.exception.SyntaxError
189 codepoint = int(c) * 100 + int(c2) * 10 + int(c3)
190 if codepoint > 255:
191 raise dns.exception.SyntaxError
192 unescaped += b"%c" % (codepoint)
193 else:
194 # Note that as mentioned above, if c is a Unicode
195 # code point outside of the ASCII range, then this
196 # += is converting that code point to its UTF-8
197 # encoding and appending multiple bytes to
198 # unescaped.
199 unescaped += c.encode()
200 else:
201 unescaped += c.encode()
202 return Token(self.ttype, bytes(unescaped))
205class Tokenizer:
206 """A DNS zone file format tokenizer.
208 A token object is basically a (type, value) tuple. The valid
209 types are EOF, EOL, WHITESPACE, IDENTIFIER, QUOTED_STRING,
210 COMMENT, and DELIMITER.
212 file: The file to tokenize
214 ungotten_char: The most recently ungotten character, or None.
216 ungotten_token: The most recently ungotten token, or None.
218 multiline: The current multiline level. This value is increased
219 by one every time a '(' delimiter is read, and decreased by one every time
220 a ')' delimiter is read.
222 quoting: This variable is true if the tokenizer is currently
223 reading a quoted string.
225 eof: This variable is true if the tokenizer has encountered EOF.
227 delimiters: The current delimiter dictionary.
229 line_number: The current line number
231 filename: A filename that will be returned by the where() method.
233 idna_codec: A dns.name.IDNACodec, specifies the IDNA
234 encoder/decoder. If None, the default IDNA
235 encoder/decoder is used.
236 """
238 def __init__(
239 self,
240 f: Any = sys.stdin,
241 filename: str | None = None,
242 idna_codec: dns.name.IDNACodec | None = None,
243 ):
244 """Initialize a tokenizer instance.
246 f: The file to tokenize. The default is sys.stdin.
247 This parameter may also be a string, in which case the tokenizer
248 will take its input from the contents of the string.
250 filename: the name of the filename that the where() method
251 will return.
253 idna_codec: A dns.name.IDNACodec, specifies the IDNA
254 encoder/decoder. If None, the default IDNA
255 encoder/decoder is used.
256 """
258 if isinstance(f, str):
259 f = io.StringIO(f)
260 if filename is None:
261 filename = "<string>"
262 elif isinstance(f, bytes):
263 f = io.StringIO(f.decode())
264 if filename is None:
265 filename = "<string>"
266 else:
267 if filename is None:
268 if f is sys.stdin:
269 filename = "<stdin>"
270 else:
271 filename = "<file>"
272 self.file = f
273 self.ungotten_char: str | None = None
274 self.ungotten_token: Token | None = None
275 self.multiline = 0
276 self.quoting = False
277 self.eof = False
278 self.delimiters = _DELIMITERS
279 self.line_number = 1
280 assert filename is not None
281 self.filename = filename
282 if idna_codec is None:
283 self.idna_codec: dns.name.IDNACodec = dns.name.IDNA_DEFAULT
284 else:
285 self.idna_codec = idna_codec
287 def _get_char(self) -> str:
288 """Read a character from input."""
290 if self.ungotten_char is None:
291 if self.eof:
292 c = ""
293 else:
294 c = self.file.read(1)
295 if c == "":
296 self.eof = True
297 elif c == "\n":
298 self.line_number += 1
299 else:
300 c = self.ungotten_char
301 self.ungotten_char = None
302 return c
304 def where(self) -> tuple[str, int]:
305 """Return the current location in the input.
307 Returns a (string, int) tuple. The first item is the filename of
308 the input, the second is the current line number.
309 """
311 return (self.filename, self.line_number)
313 def _unget_char(self, c: str) -> None:
314 """Unget a character.
316 The unget buffer for characters is only one character large; it is
317 an error to try to unget a character when the unget buffer is not
318 empty.
320 c: the character to unget
321 raises UngetBufferFull: there is already an ungotten char
322 """
324 if self.ungotten_char is not None:
325 # this should never happen!
326 raise UngetBufferFull # pragma: no cover
327 self.ungotten_char = c
329 def skip_whitespace(self) -> int:
330 """Consume input until a non-whitespace character is encountered.
332 The non-whitespace character is then ungotten, and the number of
333 whitespace characters consumed is returned.
335 If the tokenizer is in multiline mode, then newlines are whitespace.
337 Returns the number of characters skipped.
338 """
340 skipped = 0
341 while True:
342 c = self._get_char()
343 if c != " " and c != "\t":
344 if (c != "\n") or not self.multiline:
345 self._unget_char(c)
346 return skipped
347 skipped += 1
349 def get(self, want_leading: bool = False, want_comment: bool = False) -> Token:
350 """Get the next token.
352 want_leading: If True, return a WHITESPACE token if the
353 first character read is whitespace. The default is False.
355 want_comment: If True, return a COMMENT token if the
356 first token read is a comment. The default is False.
358 Raises dns.exception.UnexpectedEnd: input ended prematurely
360 Raises dns.exception.SyntaxError: input was badly formed
362 Returns a Token.
363 """
365 if self.ungotten_token is not None:
366 utoken = self.ungotten_token
367 self.ungotten_token = None
368 if utoken.is_whitespace():
369 if want_leading:
370 return utoken
371 elif utoken.is_comment():
372 if want_comment:
373 return utoken
374 else:
375 return utoken
376 skipped = self.skip_whitespace()
377 if want_leading and skipped > 0:
378 return Token(WHITESPACE, " ")
379 token = ""
380 ttype = IDENTIFIER
381 has_escape = False
382 while True:
383 c = self._get_char()
384 if c == "" or c in self.delimiters:
385 if c == "" and self.quoting:
386 raise dns.exception.UnexpectedEnd
387 if token == "" and ttype != QUOTED_STRING:
388 if c == "(":
389 self.multiline += 1
390 self.skip_whitespace()
391 continue
392 elif c == ")":
393 if self.multiline <= 0:
394 raise dns.exception.SyntaxError
395 self.multiline -= 1
396 self.skip_whitespace()
397 continue
398 elif c == '"':
399 if not self.quoting:
400 self.quoting = True
401 self.delimiters = _QUOTING_DELIMITERS
402 ttype = QUOTED_STRING
403 continue
404 else:
405 self.quoting = False
406 self.delimiters = _DELIMITERS
407 self.skip_whitespace()
408 continue
409 elif c == "\n":
410 return Token(EOL, "\n")
411 elif c == ";":
412 while 1:
413 c = self._get_char()
414 if c == "\n" or c == "":
415 break
416 token += c
417 if want_comment:
418 self._unget_char(c)
419 return Token(COMMENT, token)
420 elif c == "":
421 if self.multiline:
422 raise dns.exception.SyntaxError(
423 "unbalanced parentheses"
424 )
425 return Token(EOF, comment=token)
426 elif self.multiline:
427 self.skip_whitespace()
428 token = ""
429 continue
430 else:
431 return Token(EOL, "\n", comment=token)
432 else:
433 # This code exists in case we ever want a
434 # delimiter to be returned. It never produces
435 # a token currently.
436 token = c
437 ttype = DELIMITER
438 else:
439 self._unget_char(c)
440 break
441 elif self.quoting and c == "\n":
442 raise dns.exception.SyntaxError("newline in quoted string")
443 elif c == "\\":
444 #
445 # It's an escape. Put it and the next character into
446 # the token; it will be checked later for goodness.
447 #
448 token += c
449 has_escape = True
450 c = self._get_char()
451 if c == "" or (c == "\n" and not self.quoting):
452 raise dns.exception.UnexpectedEnd
453 token += c
454 if token == "" and ttype != QUOTED_STRING:
455 if self.multiline:
456 raise dns.exception.SyntaxError("unbalanced parentheses")
457 ttype = EOF
458 return Token(ttype, token, has_escape)
460 def unget(self, token: Token) -> None:
461 """Unget a token.
463 The unget buffer for tokens is only one token large; it is
464 an error to try to unget a token when the unget buffer is not
465 empty.
467 token: the token to unget
469 Raises UngetBufferFull: there is already an ungotten token
470 """
472 if self.ungotten_token is not None:
473 raise UngetBufferFull
474 self.ungotten_token = token
476 def next(self):
477 """Return the next item in an iteration.
479 Returns a Token.
480 """
482 token = self.get()
483 if token.is_eof():
484 raise StopIteration
485 return token
487 __next__ = next
489 def __iter__(self):
490 return self
492 # Helpers
494 def get_int(self, base: int = 10) -> int:
495 """Read the next token and interpret it as an unsigned integer.
497 Raises dns.exception.SyntaxError if not an unsigned integer.
499 Returns an int.
500 """
501 return self.as_int(self.get().unescape(), base)
503 def get_uint8(self) -> int:
504 """Read the next token and interpret it as an 8-bit unsigned
505 integer.
507 Raises dns.exception.SyntaxError if not an 8-bit unsigned integer.
509 Returns an int.
510 """
512 return self.as_uint8(self.get().unescape())
514 def get_uint16(self, base: int = 10) -> int:
515 """Read the next token and interpret it as a 16-bit unsigned
516 integer.
518 Raises dns.exception.SyntaxError if not a 16-bit unsigned integer.
520 Returns an int.
521 """
523 return self.as_uint16(self.get().unescape(), base)
525 def get_uint32(self, base: int = 10) -> int:
526 """Read the next token and interpret it as a 32-bit unsigned
527 integer.
529 Raises dns.exception.SyntaxError if not a 32-bit unsigned integer.
531 Returns an int.
532 """
534 return self.as_uint32(self.get().unescape(), base)
536 def get_uint48(self, base: int = 10) -> int:
537 """Read the next token and interpret it as a 48-bit unsigned
538 integer.
540 Raises dns.exception.SyntaxError if not a 48-bit unsigned integer.
542 Returns an int.
543 """
545 return self.as_uint48(self.get().unescape(), base)
547 def get_string(self, max_length: int | None = None) -> str:
548 """Read the next token and interpret it as a string.
550 Raises dns.exception.SyntaxError if not a string.
551 Raises dns.exception.SyntaxError if token value length
552 exceeds max_length (if specified).
554 Returns a string.
555 """
557 return self.as_string(self.get().unescape(), max_length)
559 def get_bytes(self, max_length: int | None = None) -> bytes:
560 """Read the next token and interpret it as a byte string,
561 applying any DNS escapes directly to bytes.
563 Raises dns.exception.SyntaxError if not a string.
564 Raises dns.exception.SyntaxError if the byte length
565 exceeds max_length (if specified).
567 Returns bytes.
568 """
570 return self.as_bytes(self.get().unescape_to_bytes(), max_length)
572 def get_identifier(self) -> str:
573 """Read the next token, which should be an identifier.
575 Raises dns.exception.SyntaxError if not an identifier.
577 Returns a string.
578 """
580 return self.as_identifier(self.get().unescape())
582 def get_remaining(self, max_tokens: int | None = None) -> list[Token]:
583 """Return the remaining tokens on the line, until an EOL or EOF is seen.
585 max_tokens: If not None, stop after this number of tokens.
587 Returns a list of tokens.
588 """
590 tokens = []
591 while True:
592 token = self.get()
593 if token.is_eol_or_eof():
594 self.unget(token)
595 break
596 tokens.append(token)
597 if len(tokens) == max_tokens:
598 break
599 return tokens
601 def concatenate_remaining_identifiers(self, allow_empty: bool = False) -> str:
602 """Read the remaining tokens on the line, which should be identifiers.
604 Raises dns.exception.SyntaxError if there are no remaining tokens,
605 unless `allow_empty=True` is given.
607 Raises dns.exception.SyntaxError if a token is seen that is not an
608 identifier.
610 Returns a string containing a concatenation of the remaining
611 identifiers.
612 """
613 s = ""
614 while True:
615 token = self.get().unescape()
616 if token.is_eol_or_eof():
617 self.unget(token)
618 break
619 if not token.is_identifier():
620 raise dns.exception.SyntaxError
621 s += token.value
622 if not (allow_empty or s):
623 raise dns.exception.SyntaxError("expecting another identifier")
624 return s
626 def as_name(
627 self,
628 token: Token,
629 origin: dns.name.Name | None = None,
630 relativize: bool = False,
631 relativize_to: dns.name.Name | None = None,
632 ) -> dns.name.Name:
633 """Try to interpret the token as a DNS name.
635 Raises dns.exception.SyntaxError if not a name.
637 Returns a dns.name.Name.
638 """
639 if not token.is_identifier():
640 raise dns.exception.SyntaxError("expecting an identifier")
641 name = dns.name.from_text(token.value, origin, self.idna_codec)
642 return name.choose_relativity(relativize_to or origin, relativize)
644 def as_int(self, token: Token, base: int = 10) -> int:
645 """Try to interpret the token as an unsigned integer.
647 The token must consist only of ASCII digits valid in *base*, which
648 must be between 2 and 36 inclusive. A sign, whitespace, underscores,
649 and base prefixes such as ``0x`` are not allowed.
651 Raises dns.exception.SyntaxError if not an unsigned integer.
653 Returns an int.
654 """
656 if not token.is_identifier():
657 raise dns.exception.SyntaxError("expecting an identifier")
658 if not dns._text_util.is_ascii_digits(token.value, base):
659 raise dns.exception.SyntaxError("expecting an integer")
660 return int(token.value, base)
662 def as_uint8(self, token: Token) -> int:
663 """Try to interpret the token as an unsigned 8-bit integer.
665 Raises dns.exception.SyntaxError if not 8-bit unsigned integer.
667 Returns an int.
668 """
670 value = self.as_int(token=token)
671 if value < 0 or value > 255:
672 raise dns.exception.SyntaxError(f"{value} is not an unsigned 8-bit integer")
673 return value
675 def as_uint16(self, token: Token, base: int = 10) -> int:
676 """Try to interpret the token as an unsigned 16-bit integer.
678 Raises dns.exception.SyntaxError if not a 16-bit unsigned integer.
680 Returns an int.
681 """
683 value = self.as_int(token=token, base=base)
684 if value < 0 or value > 65535:
685 if base == 8:
686 raise dns.exception.SyntaxError(
687 f"{value:o} is not an octal unsigned 16-bit integer"
688 )
689 else:
690 raise dns.exception.SyntaxError(
691 f"{value} is not an unsigned 16-bit integer"
692 )
693 return value
695 def as_uint32(self, token: Token, base: int = 10) -> int:
696 """Try to interpret the token as an unsigned 32-bit integer.
698 Raises dns.exception.SyntaxError if not a 32-bit unsigned integer.
700 Returns an int.
701 """
703 value = self.as_int(token=token, base=base)
704 if value < 0 or value > 4294967295:
705 raise dns.exception.SyntaxError(
706 f"{value} is not an unsigned 32-bit integer"
707 )
708 return value
710 def as_uint48(self, token: Token, base: int = 10) -> int:
711 """Try to interpret the token as an unsigned 48-bit integer.
713 Raises dns.exception.SyntaxError if not a 48-bit unsigned integer.
715 Returns an int.
716 """
718 value = self.as_int(token=token, base=base)
719 if value < 0 or value > 281474976710655:
720 raise dns.exception.SyntaxError(
721 f"{value} is not an unsigned 48-bit integer"
722 )
723 return value
725 def as_string(self, token: Token, max_length: int | None = None) -> str:
726 """Try to interpret the token as a string.
728 Raises dns.exception.SyntaxError if not a string.
729 Raises dns.exception.SyntaxError if token value length
730 exceeds max_length (if specified).
732 Returns a string.
733 """
735 if not (token.is_identifier() or token.is_quoted_string()):
736 raise dns.exception.SyntaxError("expecting a string")
737 if max_length and len(token.value) > max_length:
738 raise dns.exception.SyntaxError("string too long")
739 return token.value
741 def as_bytes(self, token: Token, max_length: int | None = None) -> bytes:
742 """Try to interpret the token as a byte string.
744 Raises dns.exception.SyntaxError if not a string.
745 Raises dns.exception.SyntaxError if the byte length
746 exceeds max_length (if specified).
748 Returns bytes.
749 """
751 if not (token.is_identifier() or token.is_quoted_string()):
752 raise dns.exception.SyntaxError("expecting a string")
753 value = token.value
754 if isinstance(value, str):
755 value = value.encode()
756 if max_length and len(value) > max_length:
757 raise dns.exception.SyntaxError("string too long")
758 return value
760 def as_identifier(self, token: Token) -> str:
761 """Try to interpret the token as an identifier.
763 Raises dns.exception.SyntaxError if not an identifier.
765 Returns a string.
766 """
768 if not token.is_identifier():
769 raise dns.exception.SyntaxError("expecting an identifier")
770 return token.value
772 def get_name(
773 self,
774 origin: dns.name.Name | None = None,
775 relativize: bool = False,
776 relativize_to: dns.name.Name | None = None,
777 ) -> dns.name.Name:
778 """Read the next token and interpret it as a DNS name.
780 Raises dns.exception.SyntaxError if not a name.
782 Returns a dns.name.Name.
783 """
785 token = self.get()
786 return self.as_name(token, origin, relativize, relativize_to)
788 def get_eol_as_token(self) -> Token:
789 """Read the next token and raise an exception if it isn't EOL or
790 EOF.
792 Returns a string.
793 """
795 token = self.get()
796 if not token.is_eol_or_eof():
797 raise dns.exception.SyntaxError(
798 f'expected EOL or EOF, got {token.ttype} "{token.value}"'
799 )
800 return token
802 def get_eol(self) -> str:
803 return self.get_eol_as_token().value
805 def get_ttl(self) -> int:
806 """Read the next token and interpret it as a DNS TTL.
808 Raises dns.exception.SyntaxError or dns.ttl.BadTTL if not an
809 identifier or badly formed.
811 Returns an int.
812 """
814 token = self.get().unescape()
815 if not token.is_identifier():
816 raise dns.exception.SyntaxError("expecting an identifier")
817 return dns.ttl.from_text(token.value)