Used for making fake DNS resolution responses based on received raw request Reference(s): https://code.activestate.com/recipes/491264-mini-fake-dns-server/ https://web.archive.org/web/20150418152405/https://code.google.com/p/marlon-tools/source/browse/tools/dnsproxy/dns
| 63 | return retVal |
| 64 | |
| 65 | class DNSServer(object): |
| 66 | """ |
| 67 | Used for making fake DNS resolution responses based on received |
| 68 | raw request |
| 69 | |
| 70 | Reference(s): |
| 71 | https://code.activestate.com/recipes/491264-mini-fake-dns-server/ |
| 72 | https://web.archive.org/web/20150418152405/https://code.google.com/p/marlon-tools/source/browse/tools/dnsproxy/dnsproxy.py |
| 73 | """ |
| 74 | |
| 75 | def __init__(self): |
| 76 | self._check_localhost() |
| 77 | self._requests = [] |
| 78 | self._lock = threading.Lock() |
| 79 | |
| 80 | try: |
| 81 | self._socket = socket._orig_socket(socket.AF_INET, socket.SOCK_DGRAM) |
| 82 | except AttributeError: |
| 83 | self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
| 84 | |
| 85 | self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) |
| 86 | self._socket.bind(("", 53)) |
| 87 | self._running = False |
| 88 | self._initialized = False |
| 89 | |
| 90 | def _check_localhost(self): |
| 91 | response = b"" |
| 92 | |
| 93 | try: |
| 94 | s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) |
| 95 | s.connect(("", 53)) |
| 96 | s.send(binascii.unhexlify("6509012000010000000000010377777706676f6f676c6503636f6d00000100010000291000000000000000")) # A www.google.com |
| 97 | response = s.recv(512) |
| 98 | except: |
| 99 | pass |
| 100 | finally: |
| 101 | if response and b"google" in response: |
| 102 | raise socket.error("another DNS service already running on '0.0.0.0:53'") |
| 103 | |
| 104 | def pop(self, prefix=None, suffix=None): |
| 105 | """ |
| 106 | Returns received DNS resolution request (if any) that has given |
| 107 | prefix/suffix combination (e.g. prefix.<query result>.suffix.domain) |
| 108 | """ |
| 109 | |
| 110 | retVal = None |
| 111 | |
| 112 | if prefix and hasattr(prefix, "encode"): |
| 113 | prefix = prefix.encode() |
| 114 | |
| 115 | if suffix and hasattr(suffix, "encode"): |
| 116 | suffix = suffix.encode() |
| 117 | |
| 118 | with self._lock: |
| 119 | for _ in self._requests: |
| 120 | if prefix is None and suffix is None or re.search(b"%s\\..+\\.%s" % (prefix, suffix), _, re.I): |
| 121 | self._requests.remove(_) |
| 122 | retVal = _.decode() |
no outgoing calls
no test coverage detected
searching dependent graphs…