4020719131

Build # 4020719131

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #42 from ninoseki/sigma

Run Details

300 of 421 new or added lines in 19 files covered. (71.26%)

7 existing lines in 4 files now uncovered.

608 of 791 relevant lines covered (76.86%)

1.54 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

88.89

/abuse_whois/main.py

from functools import partial

import aiometer

from abuse_whois.matchers.shared_hosting import get_shared_hosting_provider
from abuse_whois.matchers.whois import get_contact_from_whois

from .errors import InvalidAddressError, TimeoutError
from .ip import resolve_ip_address
from .schemas import Contact, Contacts, WhoisRecord
from .utils import (
    get_hostname,
    get_registered_domain,
    is_domain,
    is_ip_address,
    is_supported_address,
)
from .whois import get_whois_record as _get_whois_record


async def get_whois_record(hostname: str) -> WhoisRecord | None:
    try:
        return await _get_whois_record(hostname)
    except TimeoutError:
        return None


async def get_contact(domain_or_ip_address: str | None = None) -> Contact | None:
    if domain_or_ip_address is None:
        return None

    return await get_contact_from_whois(domain_or_ip_address)


async def get_registrar_and_hosting_provider_contacts(
    *, domain: str | None = None, ip_address: str | None = None
):
    values = [domain, ip_address]
    return await aiometer.run_all([partial(get_contact, value) for value in values])


async def get_abuse_contacts(address: str) -> Contacts:
    if not is_supported_address(address):
        raise InvalidAddressError(f"{address} is not supported type address")

    shared_hosting_provider: Contact | None = None
    registrar: Contact | None = None
    hosting_provider: Contact | None = None

    hostname = get_hostname(address)

    domain: str | None = None  # FQDN
    ip_address: str | None = None
    registered_domain: str | None = None
    whois_record: WhoisRecord | None = None

    shared_hosting_provider = get_shared_hosting_provider(hostname)

    if is_domain(hostname):
        domain = hostname
        registered_domain = get_registered_domain(hostname)

        # get IP address by domain
        try:
            ip_address = await resolve_ip_address(hostname)
        except OSError:
            pass

    if is_ip_address(hostname):
        ip_address = hostname

    registrar, hosting_provider = await get_registrar_and_hosting_provider_contacts(
        domain=domain, ip_address=ip_address
    )

    # it will get cached result (if there is no TimeoutError)
    whois_record = await get_whois_record(hostname)

    return Contacts(
        address=address,
        hostname=hostname,
        ip_address=ip_address,
        registered_domain=registered_domain,
        shared_hosting_provider=shared_hosting_provider,
        registrar=registrar,
        hosting_provider=hosting_provider,
        whois_record=whois_record,
    )

1	from functools import partial	2✔
2
3	import aiometer	2✔
4
5	from abuse_whois.matchers.shared_hosting import get_shared_hosting_provider	2✔
6	from abuse_whois.matchers.whois import get_contact_from_whois	2✔
7
8	from .errors import InvalidAddressError, TimeoutError	2✔
9	from .ip import resolve_ip_address	2✔
10	from .schemas import Contact, Contacts, WhoisRecord	2✔
11	from .utils import (	2✔
12	get_hostname,
13	get_registered_domain,
14	is_domain,
15	is_ip_address,
16	is_supported_address,
17	)
18	from .whois import get_whois_record as _get_whois_record	2✔
19
20
21	async def get_whois_record(hostname: str) -> WhoisRecord \| None:	2✔
22	try:	2✔
23	return await _get_whois_record(hostname)	2✔
UNCOV 24	except TimeoutError:	×
25	return None	×
26
27
28	async def get_contact(domain_or_ip_address: str \| None = None) -> Contact \| None:	2✔
29	if domain_or_ip_address is None:	2✔
30	return None	2✔
31
32	return await get_contact_from_whois(domain_or_ip_address)	2✔
33
34
35	async def get_registrar_and_hosting_provider_contacts(	2✔
36	*, domain: str \| None = None, ip_address: str \| None = None
37	):
38	values = [domain, ip_address]	2✔
39	return await aiometer.run_all([partial(get_contact, value) for value in values])	2✔
40
41
42	async def get_abuse_contacts(address: str) -> Contacts:	2✔
43	if not is_supported_address(address):	2✔
UNCOV 44	raise InvalidAddressError(f"{address} is not supported type address")	×
45
46	shared_hosting_provider: Contact \| None = None	2✔
47	registrar: Contact \| None = None	2✔
48	hosting_provider: Contact \| None = None	2✔
49
50	hostname = get_hostname(address)	2✔
51
52	domain: str \| None = None # FQDN	2✔
53	ip_address: str \| None = None	2✔
54	registered_domain: str \| None = None	2✔
55	whois_record: WhoisRecord \| None = None	2✔
56
57	shared_hosting_provider = get_shared_hosting_provider(hostname)	2✔
58
59	if is_domain(hostname):	2✔
60	domain = hostname	2✔
61	registered_domain = get_registered_domain(hostname)	2✔
62
63	# get IP address by domain
64	try:	2✔
65	ip_address = await resolve_ip_address(hostname)	2✔
UNCOV 66	except OSError:	×
67	pass	×
68
69	if is_ip_address(hostname):	2✔
70	ip_address = hostname	2✔
71
72	registrar, hosting_provider = await get_registrar_and_hosting_provider_contacts(	2✔
73	domain=domain, ip_address=ip_address
74	)
75
76	# it will get cached result (if there is no TimeoutError)
77	whois_record = await get_whois_record(hostname)	2✔
78
79	return Contacts(	2✔
80	address=address,
81	hostname=hostname,
82	ip_address=ip_address,
83	registered_domain=registered_domain,
84	shared_hosting_provider=shared_hosting_provider,
85	registrar=registrar,
86	hosting_provider=hosting_provider,
87	whois_record=whois_record,
88	)

ninoseki / abuse_whois / 4020719131

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous