File: util.py

package info (click to toggle)
dnspython 2.7.0-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 2,448 kB
  • sloc: python: 34,885; sh: 7; makefile: 4
file content (133 lines) | stat: -rw-r--r-- 4,480 bytes parent folder | download
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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license

# Copyright (C) 2003-2007, 2009-2011 Nominum, Inc.
#
# Permission to use, copy, modify, and distribute this software and its
# documentation for any purpose with or without fee is hereby granted,
# provided that the above copyright notice and this permission notice
# appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

import enum
import inspect
import os

import dns.message
import dns.name
import dns.query
import dns.rdataclass
import dns.rdatatype

# Cache for is_internet_reachable()
_internet_reachable = None
_have_ipv4 = False
_have_ipv6 = False


def here(filename):
    return os.path.join(os.path.dirname(__file__), filename)


def check_networking(addresses):
    """Can we do a DNS resolution via UDP and TCP to at least one of the addresses?"""
    for address in addresses:
        try:
            q = dns.message.make_query(dns.name.root, dns.rdatatype.NS)
            ok = False
            # We try UDP a few times in case we get unlucky and a packet is lost.
            for i in range(5):
                # We don't check the answer other than make sure there is one.
                try:
                    r = dns.query.udp(q, address, timeout=4)
                    ns = r.find_rrset(
                        r.answer, dns.name.root, dns.rdataclass.IN, dns.rdatatype.NS
                    )
                    ok = True
                    break
                except Exception:
                    continue  # UDP try loop
            if not ok:
                continue  # addresses loop
            try:
                r = dns.query.tcp(q, address, timeout=4)
                ns = r.find_rrset(
                    r.answer, dns.name.root, dns.rdataclass.IN, dns.rdatatype.NS
                )
                # UDP and TCP both work!
                return True
            except Exception:
                continue
        except Exception as e:
            pass
    return False


def is_internet_reachable():
    """Check if the Internet is reachable.

    Setting the environment variable `NO_INTERNET` will let this
    function always return False. The result is cached.

    We check using the Google and Cloudflare public resolvers as they are highly
    available and have well-known stable addresses.
    """
    global _internet_reachable
    if _internet_reachable is None:
        if os.environ.get("NO_INTERNET"):
            _internet_reachable = False
        else:
            global _have_ipv4
            _have_ipv4 = check_networking(["8.8.8.8", "1.1.1.1"])
            global _have_ipv6
            _have_ipv6 = check_networking(
                ["2001:4860:4860::8888", "2606:4700:4700::1111"]
            )
            _internet_reachable = _have_ipv4 or _have_ipv6
    return _internet_reachable


def have_ipv4():
    if not is_internet_reachable():
        return False
    return _have_ipv4


def have_ipv6():
    if not is_internet_reachable():
        return False
    return _have_ipv6


def enumerate_module(module, super_class):
    """Yield module attributes which are subclasses of given class"""
    for attr_name in dir(module):
        attr = getattr(module, attr_name)
        if inspect.isclass(attr) and issubclass(attr, super_class):
            yield attr


def check_enum_exports(module, eq_callback, only=None):
    """Make sure module exports all mnemonics from enums"""
    for attr in enumerate_module(module, enum.Enum):
        if only is not None and attr not in only:
            # print('SKIP', attr)
            continue
        for flag, value in attr.__members__.items():
            # print(module, flag, value)
            eq_callback(getattr(module, flag), value)


def is_docker() -> bool:
    # There are a lot of answers to "am I runnning in a container" and none appear
    # to work reliably, so we're settling for "am I running in docker?"
    try:
        return os.path.isfile("/.dockerenv")
    except Exception:
        return False