"""Enables interacting with the DomainTools API via Python:

limits = {}

limits_set = False

def __init__(self, username, key, https=True, verify_ssl=True, rate_limit=True, proxy_url=None,

always_sign_api_key=False, key_sign_hash='sha256', app_name="python_wrapper", app_version=version,

**default_parameters):

if not default_parameters:

self.default_parameters = {}

else:

self.default_parameters = default_parameters

self.username = username

self.key = key

self.https = https

self.verify_ssl = verify_ssl

self.rate_limit = rate_limit

self.proxy_url = proxy_url

self.extra_request_params = {}

self.always_sign_api_key = always_sign_api_key

self.key_sign_hash = key_sign_hash

self.default_parameters['app_name'] = app_name

self.default_parameters['app_version'] = app_version

if not https:

raise Exception("The DomainTools API endpoints no longer support http traffic. Please make sure https=True.")

if proxy_url:

if isinstance(proxy_url, str):

self.proxy_url = {'http://': proxy_url, 'https://': proxy_url}

else:

raise Exception("Proxy URL must be a string. For example: '127.0.0.1:8888'")

def _rate_limit(self):

"""Pulls in and enforces the latest rate limits for the specified user"""

self.limits_set = True

for product in self.account_information():

limit_minutes = product['per_minute_limit'] or None

limit_hours = product['per_hour_limit'] or None

default = 3600

hours = limit_hours and 3600 / float(limit_hours)

minutes = limit_minutes and 60 / float(limit_minutes)

self.limits[product['id']] = {'interval': timedelta(

seconds = minutes or hours or default

)

}

def _results(self, product, path, cls=Results, **kwargs):

"""Returns _results for the specified API path with the specified **kwargs parameters"""

if product != 'account-information' and self.rate_limit and not self.limits_set and not self.limits:

self._rate_limit()

uri = '/'.join(('https://api.domaintools.com', path.lstrip('/')))

parameters = self.default_parameters.copy()

parameters['api_username'] = self.username

self.handle_api_key(path, parameters)

parameters.update(

{

key: str(value).lower() if value in (True, False) else value

for key, value in kwargs.items()

if value is not None

}

)

return cls(self, product, uri, **parameters)

def handle_api_key(self, path, parameters):

if self.https and not self.always_sign_api_key:

parameters['api_key'] = self.key

else:

if self.key_sign_hash and self.key_sign_hash in AVAILABLE_KEY_SIGN_HASHES:

signing_hash = eval(self.key_sign_hash)

else:

raise ValueError("Invalid value '{0}' for 'key_sign_hash'. "

"Values available are {1}".format(self.key_sign_hash, ','.join(AVAILABLE_KEY_SIGN_HASHES)))

parameters['timestamp'] = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ')

parameters['signature'] = hmac(self.key.encode('utf8'),

''.join([self.username, parameters['timestamp'], path]).encode('utf8'),

digestmod=signing_hash).hexdigest()

def account_information(self, **kwargs):

"""Provides a snapshot of your accounts current API usage"""

return self._results('account-information', '/v1/account', items_path=('products',), **kwargs)

def available_api_calls(self):

"""Provides a list of api calls that you can use based on your account information."""

def snakecase(string):

string = re.sub(r"[\-\.\s]", '_', str(string))

if not string:

return string

return str(string[0]).lower() + re.sub(r"[A-Z]", lambda matched: '_' + str(matched.group(0)).lower(),

string[1:])

api_calls = tuple((api_call for api_call in dir(API) if not api_call.startswith('_') and

callable(getattr(API, api_call, None))))

return sorted([snakecase(p["id"]) for p in self.account_information()["products"] if snakecase(p["id"]) in api_calls])

def brand_monitor(self, query, exclude=None, domain_status=None, days_back=None, **kwargs):

"""Pass in one or more terms as a list or separated by the pipe character ( | )"""

if exclude is None:

exclude = []

return self._results('mark-alert', '/v1/mark-alert', query=delimited(query), exclude=delimited(exclude),

domain_status=domain_status, days_back=days_back, items_path=('alerts',), **kwargs)

def domain_profile(self, query, **kwargs):

"""Returns a profile for the specified domain name"""

return self._results('domain-profile', '/v1/{0}'.format(query))

def domain_search(self, query, exclude_query=None, max_length=25, min_length=2, has_hyphen=True, has_number=True,

active_only=False, deleted_only=False, anchor_left=False, anchor_right=False, page=1, **kwargs):

"""Each term in the query string must be at least three characters long.

if exclude_query is None:

exclude_query = []

return self._results('domain-search', '/v2/domain-search', query=delimited(query, ' '),

exclude_query=delimited(exclude_query, ' '),

max_length=max_length, min_length=min_length, has_hyphen=has_hyphen, has_number=has_number,

active_only=active_only, deleted_only=deleted_only, anchor_left=anchor_left,

anchor_right=anchor_right, page=page, items_path=('results',), **kwargs)

def hosting_history(self, query, **kwargs):

"""Returns the hosting history from the given domain name"""

return self._results('hosting-history', '/v1/{0}/hosting-history'.format(query), cls=GroupedIterable, **kwargs)

def ip_monitor(self, query, days_back=0, page=1, **kwargs):

"""Pass in the IP Address you wish to query ( i.e. 199.30.228.112 )."""

return self._results('ip-monitor', '/v1/ip-monitor', query=query, days_back=days_back, page=page,

items_path=('alerts',), **kwargs)

def ip_registrant_monitor(self, query, days_back=0, search_type="all", server=None, country=None, org=None, page=1,

include_total_count=False, **kwargs):

"""Query based on free text query terms"""

return self._results('ip-registrant-monitor', '/v1/ip-registrant-monitor', query=query,

days_back=days_back, search_type=search_type, server=server, country=country, org=org,

page=page, include_total_count=include_total_count, **kwargs)

def name_server_monitor(self, query, days_back=0, page=1, **kwargs):

"""Pass in the hostname of the Name Server you wish to query ( i.e. dynect.net )."""

return self._results('name-server-monitor', '/v1/name-server-monitor', query=query, days_back=days_back,

page=page, items_path=('alerts',), **kwargs)

def parsed_whois(self, query, **kwargs):

"""Pass in a domain name"""

return self._results('parsed-whois', '/v1/{0}/whois/parsed'.format(query), cls=ParsedWhois, **kwargs)

def registrant_monitor(self, query, exclude=None, days_back=0, limit=None, **kwargs):

"""One or more terms as a Python list or separated by the pipe character ( | )."""

if exclude is None:

exclude = []

return self._results('registrant-alert', '/v1/registrant-alert', query=delimited(query),

exclude=delimited(exclude), days_back=days_back, limit=limit, items_path=('alerts',),

**kwargs)

def reputation(self, query, include_reasons=False, **kwargs):

"""Pass in a domain name to see its reputation score"""

return self._results('reputation', '/v1/reputation', domain=query, include_reasons=include_reasons,

cls=Reputation, **kwargs)

def reverse_ip(self, domain=None, limit=None, **kwargs):

"""Pass in a domain name."""

return self._results('reverse-ip', '/v1/{0}/reverse-ip'.format(domain), limit=limit, **kwargs)

def host_domains(self, ip=None, limit=None, **kwargs):

"""Pass in an IP address."""

return self._results('reverse-ip', '/v1/{0}/host-domains'.format(ip), limit=limit, **kwargs)

def reverse_ip_whois(self, query=None, ip=None, country=None, server=None, include_total_count=False, page=1,

**kwargs):

"""Pass in an IP address or a list of free text query terms."""

if (ip and query) or not (ip or query):

raise ValueError('Query or IP Address (but not both) must be defined')

return self._results('reverse-ip-whois', '/v1/reverse-ip-whois', query=query, ip=ip, country=country,

server=server, include_total_count=include_total_count, page=page, items_path=('records',),

**kwargs)

def reverse_name_server(self, query, limit=None, **kwargs):

"""Pass in a domain name or a name server."""

return self._results('reverse-name-server', '/v1/{0}/name-server-domains'.format(query),

items_path=('primary_domains',), limit=limit, **kwargs)

def reverse_whois(self, query, exclude=None, scope='current', mode='purchase', **kwargs):

"""List of one or more terms to search for in the Whois record,

if exclude is None:

exclude = []

return self._results('reverse-whois', '/v1/reverse-whois', terms=delimited(query), exclude=delimited(exclude),

scope=scope, mode=mode, **kwargs)

def whois(self, query, **kwargs):

"""Pass in a domain name or an IP address to perform a whois lookup."""

return self._results('whois', '/v1/{0}/whois'.format(query), **kwargs)

def whois_history(self, query, mode=None, sort=None, offset=None, limit=None, **kwargs):

"""Pass in a domain name."""

return self._results('whois-history', '/v1/{0}/whois/history'.format(query), mode=mode, sort=sort,

offset=offset, limit=limit, items_path=('history',), **kwargs)

def phisheye(self, query, days_back=None, **kwargs):

"""Returns domain results for the specified term for today or the specified number of days_back.

return self._results('phisheye', '/v1/phisheye', query=query, days_back=days_back, items_path=('domains',),

**kwargs)

def phisheye_term_list(self, include_inactive=False, **kwargs):

"""Provides a list of terms that are set up for this account.

return self._results('phisheye_term_list', '/v1/phisheye/term-list', include_inactive=include_inactive,

items_path=('terms',), **kwargs)

def iris(self, domain=None, ip=None, email=None, nameserver=None, registrar=None, registrant=None,

registrant_org=None, **kwargs):

"""Performs a search for the provided search terms ANDed together,

if ((not domain and not ip and not email and not nameserver and not registrar and not registrant and not

registrant_org and not kwargs)):

raise ValueError('At least one search term must be specified')

return self._results('iris', '/v1/iris', domain=domain, ip=ip, email=email, nameserver=nameserver,

registrar=registrar, registrant=registrant, registrant_org=registrant_org,

items_path=('results',), **kwargs)

def risk(self, domain, **kwargs):

"""Returns back the risk score for a given domain"""

return self._results('risk', '/v1/risk', items_path=('components',), domain=domain, cls=Reputation,

**kwargs)

def risk_evidence(self, domain, **kwargs):

"""Returns back the detailed risk evidence associated with a given domain"""

return self._results('risk-evidence', '/v1/risk/evidence/', items_path=('components',), domain=domain,

**kwargs)

def iris_enrich(self, *domains, **kwargs):

"""Returns back enriched data related to the specified domains using our Iris Enrich service

if not domains:

raise ValueError('One or more domains to enrich must be provided')

domains = ','.join(domains)

data_updated_after = kwargs.get('data_updated_after', None)

if hasattr(data_updated_after, 'strftime'):

data_updated_after = data_updated_after.strftime('%Y-%m-%d')

return self._results('iris-enrich', '/v1/iris-enrich/', domain=domains, data_updated_after=data_updated_after,

items_path=('results',), **kwargs)

def iris_enrich_cli(self, domains=None, **kwargs):

"""Returns back enriched data related to the specified domains using our Iris Enrich service.

if not domains:

raise ValueError('One or more domains to enrich must be provided')

if isinstance(domains, (list, tuple)):

domains = ','.join(domains)

data_updated_after = kwargs.get('data_updated_after', None)

if hasattr(data_updated_after, 'strftime'):

data_updated_after = data_updated_after.strftime('%Y-%m-%d')

return self._results('iris-enrich', '/v1/iris-enrich/', domain=domains, data_updated_after=data_updated_after,

items_path=('results',), **kwargs)

def iris_investigate(self, domains=None, data_updated_after=None, expiration_date=None,

create_date=None, active=None, search_hash=None, **kwargs):

"""Returns back a list of domains based on the provided filters.

# We put search_hash in the signature definition so the CLI can see it as a valid arg

if search_hash:

kwargs['search_hash'] = search_hash

if not (kwargs or domains):

raise ValueError('Need to define investigation using kwarg filters or domains')

if isinstance(domains, (list, tuple)):

domains = ','.join(domains)

if hasattr(data_updated_after, 'strftime'):

data_updated_after = data_updated_after.strftime('%Y-%m-%d')

if hasattr(expiration_date, 'strftime'):

expiration_date = expiration_date.strftime('%Y-%m-%d')

if hasattr(create_date, 'strftime'):

create_date = create_date.strftime('%Y-%m-%d')

if isinstance(active, bool):

kwargs['active'] = str(active).lower()

return self._results('iris-investigate', '/v1/iris-investigate/', domain=domains,

data_updated_after=data_updated_after, expiration_date=expiration_date,

create_date=create_date, items_path=('results',), **kwargs)

def iris_detect_monitors(self, include_counts=False, datetime_counts_since=None, sort=None, order="desc", offset=0,

limit=None, **kwargs):

"""Returns back a list of monitors in Iris Detect based on the provided filters.

if include_counts:

if not datetime_counts_since:

raise ValueError('Need to define datetime_counts_since when include_counts is True')

if isinstance(datetime_counts_since, datetime):

datetime_counts_since = str(datetime_counts_since.astimezone())

elif isinstance(datetime_counts_since, str):

kwargs["datetime_counts_since"] = datetime_counts_since

kwargs["include_counts"] = "true"

kwargs["datetime_counts_since"] = datetime_counts_since

if sort:

kwargs["sort[]"] = sort

return self._results('iris-detect-monitors', '/v1/iris-detect/monitors/', order=order, offset=offset,

limit=limit, items_path=('monitors',), response_path=(), **kwargs)

def iris_detect_new_domains(self, monitor_id=None, tlds=None, risk_score_ranges=None,

mx_exists=None, discovered_since=None, changed_since=None, search=None, sort=None,

order=None, include_domain_data=False, offset=0, limit=None, preview=None, **kwargs):

"""Returns back a list of new domains in Iris Detect based on the provided filters.

if discovered_since:

if isinstance(discovered_since, datetime):

kwargs["discovered_since"] = str(discovered_since.astimezone())

elif isinstance(discovered_since, str):

kwargs["discovered_since"] = discovered_since

if changed_since:

if isinstance(changed_since, datetime):

kwargs["changed_since"] = str(changed_since.astimezone())

elif isinstance(changed_since, str):

kwargs["changed_since"] = changed_since

if tlds:

kwargs["tlds[]"] = tlds

if risk_score_ranges:

kwargs["risk_score_ranges[]"] = risk_score_ranges

if sort:

kwargs["sort[]"] = sort

if order is not None:

kwargs["order"] = order

if mx_exists is not None:

kwargs["mx_exists"] = mx_exists

return self._results('iris-detect-new-domains', '/v1/iris-detect/domains/new/', monitor_id=monitor_id,

search=search, include_domain_data=include_domain_data, preview=preview, offset=offset,

limit=limit, items_path=('watchlist_domains',), response_path=(), **kwargs)

def iris_detect_watched_domains(self, monitor_id=None, escalation_types=None, tlds=None, risk_score_ranges=None,

mx_exists=None, discovered_since=None, changed_since=None, escalated_since=None, search=None, sort=None,

order=None, include_domain_data=False, offset=0, limit=None, preview=None, **kwargs):

"""Returns back a list of watched domains in Iris Detect based on the provided filters.

if discovered_since:

if isinstance(discovered_since, datetime):

kwargs["discovered_since"] = str(discovered_since.astimezone())

elif isinstance(discovered_since, str):

kwargs["discovered_since"] = discovered_since

if changed_since:

if isinstance(changed_since, datetime):

kwargs["changed_since"] = str(changed_since.astimezone())

elif isinstance(changed_since, str):

kwargs["changed_since"] = changed_since

if escalated_since:

if isinstance(escalated_since, datetime):

kwargs["escalated_since"] = str(escalated_since.astimezone())

elif isinstance(escalated_since, str):

kwargs["escalated_since"] = escalated_since

if escalation_types:

kwargs["escalation_types[]"] = escalation_types

if tlds:

kwargs["tlds[]"] = tlds

if risk_score_ranges:

kwargs["risk_score_ranges[]"] = risk_score_ranges

if sort:

kwargs["sort[]"] = sort

if order is not None:

kwargs["order"] = order

if mx_exists is not None:

kwargs["mx_exists"] = mx_exists

return self._results('iris-detect-watched-domains', '/v1/iris-detect/domains/watched/', monitor_id=monitor_id,

search=search, include_domain_data=include_domain_data, preview=preview, offset=offset,

limit=limit, items_path=('watchlist_domains',), response_path=(), **kwargs)

def iris_detect_manage_watchlist_domains(self, watchlist_domain_ids, state, **kwargs):

"""Changes the watch state of a list of domains by their Iris Detect domain ID.

if isinstance(watchlist_domain_ids, str):

watchlist_domain_ids = watchlist_domain_ids.split(',')

return self._results('iris-detect-manage-watchlist-domains', '/v1/iris-detect/domains/', state=state,

watchlist_domain_ids=watchlist_domain_ids, items_path=('watchlist_domains',),

response_path=(), **kwargs)

def iris_detect_escalate_domains(self, watchlist_domain_ids, escalation_type, **kwargs):

"""Changes the escalation type of a list of domains by their Iris Detect domain ID.

kwargs["watchlist_domain_ids[]"] = watchlist_domain_ids

return self._results('iris-detect-escalate-domains', '/v1/iris-detect/escalations/',

escalation_type=escalation_type, items_path=('escalations',),

response_path=(), **kwargs)

def iris_detect_ignored_domains(self, monitor_id=None, escalation_types=None, tlds=None, risk_score_ranges=None,

mx_exists=None, discovered_since=None, changed_since=None, escalated_since=None, search=None, sort=None,

order=None, include_domain_data=False, offset=0, limit=None, preview=None, **kwargs):

"""Returns back a list of ignored domains in Iris Detect based on the provided filters.

if discovered_since:

if isinstance(discovered_since, datetime):

kwargs["discovered_since"] = str(discovered_since.astimezone())

elif isinstance(discovered_since, str):

kwargs["discovered_since"] = discovered_since

if changed_since:

if isinstance(changed_since, datetime):

kwargs["changed_since"] = str(changed_since.astimezone())

elif isinstance(changed_since, str):

kwargs["changed_since"] = changed_since

if escalated_since:

if isinstance(escalated_since, datetime):

kwargs["escalated_since"] = str(escalated_since.astimezone())

elif isinstance(escalated_since, str):

kwargs["escalated_since"] = escalated_since

if escalation_types:

kwargs["escalation_types[]"] = escalation_types

if tlds:

kwargs["tlds[]"] = tlds

if risk_score_ranges:

kwargs["risk_score_ranges[]"] = risk_score_ranges

if sort:

kwargs["sort[]"] = sort

if order is not None:

kwargs["order"] = order

if mx_exists is not None:

kwargs["mx_exists"] = mx_exists

return self._results('iris-detect-ignored-domains', '/v1/iris-detect/domains/ignored/', monitor_id=monitor_id,

search=search, include_domain_data=include_domain_data, preview=preview, offset=offset,