respodns/respodns/dns.py

from .structs import Options


def detect_gfw(r, ip, check):
    # attempt to detect interference from the Great Firewall of China.
    from .ips import gfw_ips

    def rs(prefix):
        return r.startswith(prefix)

    def de(suffix):
        return check.domain.endswith(suffix)

    hosted = de("facebook.com") or de("instagram.com") or de("whatsapp.com")

    if rs("31.13.") and not hosted:
        return True

    if rs("69.171.") and not rs("69.171.250."):
        return True

    if rs("108.160."):
        return True

    if r in gfw_ips:
        return True

    return False


async def getaddrs(server, domain, opts):
    from .ip_util import ipkey
    from dns.asyncresolver import Resolver
    from dns.exception import Timeout
    from dns.resolver import NXDOMAIN, NoAnswer, NoNameservers

    res = Resolver(configure=False)
    if opts.impatient:
        res.timeout = 5
        res.lifetime = 2
    else:
        res.timeout = 3
        res.lifetime = 9
    res.nameservers = [server]
    try:
        ans = await res.resolve(domain, "A", search=False)
    except NXDOMAIN:
        return ["NXDOMAIN"]
    except NoAnswer:
        return ["NoAnswer"]
    except NoNameservers:
        return ["NoNameservers"]
    except Timeout:
        return ["Timeout"]
    return sorted(set(rr.address for rr in ans.rrset), key=ipkey)


def process_result(res, ip, check, opts: Options):
    from .ips import is_bogon, is_block_target
    from .util import right_now
    from .structs import Entry

    # TODO: get more accurate times by inserting start-end into getaddrs.
    now = right_now()
    assert len(res) > 0
    reason = None

    if "Timeout" in res:
        reason = "timeout"

    elif check.kind.startswith("bad"):
        reason = "okay" if "NXDOMAIN" in res else "redirect"

    elif any(is_bogon(r) or is_block_target(r) for r in res):
        reason = "block"

    elif not any(len(r) > 0 and r[0].isdigit() for r in res):
        # TODO: check for no alias on common.
        reason = "missing"

    else:
        for r in res:
            if len(r) == 0 or not r[0].isdigit():
                continue
            if detect_gfw(r, ip, check):
                reason = "gfw"
                break
        else:
            reason = "okay"

    assert reason is not None, (res, ip, check)

    addrs = list(filter(lambda r: len(r) > 0 and r[0].isdigit(), res))
    exception = res[0] if len(addrs) == 0 else None

    return Entry(
        date=now,
        success=reason == "okay",
        server=ip,
        kind=check.kind,
        domain=check.domain,
        exception=exception,
        addrs=addrs,
        reason=reason,
        execution=opts.execution,
    )


async def try_ip(db, server_ip, checks, opts: Options):
    from .util import make_pooler
    from asyncio import sleep

    entries = []

    success = True

    def finisher(done, pending):
        nonlocal success
        for task in done:
            res, ip, check = task.result()
            entry = process_result(res, ip, check, opts)
            entries.append(entry)
            if not entry.success:
                if opts.early_stopping and success:  # only cancel once
                    for pend in pending:
                        # FIXME: this can still, somehow,
                        #        cancel the main function.
                        pend.cancel()
                success = False

    pooler = make_pooler(opts.domain_simul, finisher)

    async def getaddrs_wrapper(ip, check):
        # NOTE: could put right_now() stuff here!
        # TODO: add duration field given in milliseconds (integer)
        #       by subtracting start and end datetimes.
        res = await getaddrs(ip, check.domain, opts)
        return res, ip, check

    for i, check in enumerate(checks):
        first = i == 0
        if not first:
            await sleep(opts.domain_wait)
        await pooler(getaddrs_wrapper(server_ip, check))
        if first:
            # limit to one connection for the first check.
            await pooler()
        if not success:
            if opts.early_stopping or first:
                break
    else:
        await pooler()

    if opts.ipinfo is not None:
        await opts.ipinfo.find_country(server_ip, db)
        for entry in entries:
            for addr in entry.addrs:
                await opts.ipinfo.find_country(addr, db)
        opts.ipinfo.flush()

    if not opts.dry:
        for entry in entries:
            db.push_entry(entry)
        db.commit()

    if not success:
        first_failure = None
        assert len(entries) > 0
        for entry in entries:
            if not entry.success:
                first_failure = entry
                break
        else:
            assert 0, ("no failures found:", entries)
        return server_ip, first_failure
    return server_ip, None


async def sync_database(db, opts: Options):
    from .ips import china, blocks

    if db is None:
        return

    # TODO: handle addresses that were removed from respodns.ips.china.
    for ips, kw in ((china, "china"), (blocks, "block_target")):
        for ip in ips:
            kwargs = dict()
            kwargs[kw] = True
            if opts.ipinfo is not None:
                kwargs["country_code"] = await opts.ipinfo.find_country(ip)
            db.modify_address(ip, **kwargs)
    if opts.ipinfo is not None:
        opts.ipinfo.flush()


async def main(db, filepath, checks, opts: Options):
    from .ip_util import read_ips
    from .util import make_pooler
    from asyncio import sleep, create_task
    from sys import stdin, stderr

    syncing = create_task(sync_database(db, opts))

    def finisher(done, pending):
        for task in done:
            ip, first_failure = task.result()
            if first_failure is None:
                print(ip)
            elif opts.dry:
                ff = first_failure
                if ff.kind in ("shock", "adware"):
                    print(ip, ff.reason, ff.kind, sep="\t")
                else:
                    print(ip, ff.reason, ff.kind, ff.domain, sep="\t")

    pooler = make_pooler(opts.ip_simul, finisher)

    f = stdin if filepath == "" else open(filepath, "r")
    for i, ip in enumerate(read_ips(f)):
        first = i == 0
        if opts.progress:
            print(f"#{i}: {ip}", file=stderr)
            stderr.flush()
        if not first:
            await sleep(opts.ip_wait)
        await pooler(try_ip(db, ip, checks, opts))
    if f != stdin:
        f.close()

    await pooler()
    await syncing