/usr/lib/python2.7/dist-packages/ripe/atlas/tools/renderers/dns.py is in ripe-atlas-tools 1.2.2-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | # Copyright (c) 2015 RIPE NCC
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from tzlocal import get_localzone
from ..helpers.colours import colourise
from ..helpers.sanitisers import sanitise
from .base import Renderer as BaseRenderer
class Renderer(BaseRenderer):
RENDERS = [BaseRenderer.TYPE_DNS]
TIME_FORMAT = "%a %b %d %H:%M:%S %Z %Y"
def on_result(self, result):
created = result.created.astimezone(get_localzone())
probe_id = result.probe_id
r = "\n\nProbe #{0}\n{1}\n".format(probe_id, "=" * 79)
if result.responses:
for response in result.responses:
r += self.get_formatted_response(probe_id, created, response)
else:
r += "\n {}\n".format(colourise("No response found", "red"))
return r
@classmethod
def get_formatted_response(cls, probe_id, created, response):
if not response.abuf:
return "\n- {0} -\n\n No abuf found.\n".format(
response.response_id)
header_flags = []
for flag in ("aa", "ad", "cd", "qr", "ra", "rd",):
if getattr(response.abuf.header, flag):
header_flags.append(flag)
edns = ""
if response.abuf.edns0:
edns = "\n ;; OPT PSEUDOSECTION:\n ; EDNS: version: {0}, " \
"flags:; udp: {1}\n".format(
response.abuf.edns0.version,
response.abuf.edns0.udp_size
)
question = ""
if response.abuf.questions:
question = response.abuf.questions[0].name
return cls._colourise_by_response(response, cls.render(
"reports/dns.txt",
# Older measurements don't have a response_id
response_id=response.response_id or 1,
probe=probe_id,
question_name=sanitise(question),
header_opcode=response.abuf.header.opcode,
header_return_code=response.abuf.header.return_code,
header_id=response.abuf.header.id,
header_flags=" ".join(header_flags),
edns=edns,
question_count=len(response.abuf.questions),
answer_count=len(response.abuf.answers),
authority_count=len(response.abuf.authorities),
additional_count=len(response.abuf.additionals),
question=sanitise(cls.get_section(
"question", response.abuf.questions), strip_newlines=False),
answers=sanitise(cls.get_section(
"answer", response.abuf.answers), strip_newlines=False),
authorities=sanitise(cls.get_section(
"authority", response.abuf.authorities), strip_newlines=False),
additionals=sanitise(cls.get_section(
"additional", response.abuf.additionals), strip_newlines=False),
response_time=response.response_time,
response_size=response.response_size,
created=created.strftime(cls.TIME_FORMAT),
destination_address=sanitise(response.destination_address),
))
@staticmethod
def get_section(header, data):
if not data:
return ""
return "\n ;; {0} SECTION:\n{1}\n".format(
header.upper(),
"\n".join([" {0}".format(_) for _ in data])
)
@staticmethod
def _colourise_by_response(response, output):
colour = "red" if response.is_error else "green"
return colourise(output, colour)
|