From a04d33fbba72a11fa4b5a957400a1f3d5f549bee Mon Sep 17 00:00:00 2001 From: Nicholas McDonnell <50747025+mcdonnnj@users.noreply.github.com> Date: Tue, 14 Jul 2020 12:34:10 -0400 Subject: [PATCH 1/8] Add pylint testing to serve as a foundation for cleanup Add pylint as a testing requirement in setup.py, add a local hook using it to pre-commit-config.yaml, and add a slightly tweaked .pylintrc file for configuration. --- .pre-commit-config.yaml | 7 + .pylintrc | 597 ++++++++++++++++++++++++++++++++++++++++ setup.py | 1 + 3 files changed, 605 insertions(+) create mode 100644 .pylintrc diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 08b0935..79ac1ff 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -115,3 +115,10 @@ repos: rev: v0.782 hooks: - id: mypy + - repo: local + hooks: + - id: pylint + name: pylint + entry: pylint + language: system + types: [python] diff --git a/.pylintrc b/.pylintrc new file mode 100644 index 0000000..d56e2cd --- /dev/null +++ b/.pylintrc @@ -0,0 +1,597 @@ +[MASTER] + +# A comma-separated list of package or module names from where C extensions may +# be loaded. Extensions are loading into the active Python interpreter and may +# run arbitrary code. +extension-pkg-whitelist= + +# Specify a score threshold to be exceeded before program exits with error. +fail-under=10 + +# Add files or directories to the blacklist. They should be base names, not +# paths. +ignore=CVS + +# Add files or directories matching the regex patterns to the blacklist. The +# regex matches against base names, not paths. +ignore-patterns= + +# Python code to execute, usually for sys.path manipulation such as +# pygtk.require(). +#init-hook= + +# Use multiple processes to speed up Pylint. Specifying 0 will auto-detect the +# number of processors available to use. +jobs=1 + +# Control the amount of potential inferred values when inferring a single +# object. This can help the performance when dealing with large functions or +# complex, nested conditions. +limit-inference-results=100 + +# List of plugins (as comma separated values of python module names) to load, +# usually to register additional checkers. +load-plugins= + +# Pickle collected data for later comparisons. +persistent=yes + +# When enabled, pylint would attempt to guess common misconfiguration and emit +# user-friendly hints instead of false-positive error messages. +suggestion-mode=yes + +# Allow loading of arbitrary C extensions. Extensions are imported into the +# active Python interpreter and may run arbitrary code. +unsafe-load-any-extension=no + + +[MESSAGES CONTROL] + +# Only show warnings with the listed confidence levels. Leave empty to show +# all. Valid levels: HIGH, INFERENCE, INFERENCE_FAILURE, UNDEFINED. +confidence= + +# Disable the message, report, category or checker with the given id(s). You +# can either give multiple identifiers separated by comma (,) or put this +# option multiple times (only on the command line, not in the configuration +# file where it should appear only once). You can also use "--disable=all" to +# disable everything first and then reenable specific checks. For example, if +# you want to run only the similarities checker, you can use "--disable=all +# --enable=similarities". If you want to run only the classes checker, but have +# no Warning level messages displayed, use "--disable=all --enable=classes +# --disable=W". +disable=C0330, + print-statement, + parameter-unpacking, + unpacking-in-except, + old-raise-syntax, + backtick, + long-suffix, + old-ne-operator, + old-octal-literal, + import-star-module-level, + non-ascii-bytes-literal, + raw-checker-failed, + bad-inline-option, + locally-disabled, + file-ignored, + suppressed-message, + useless-suppression, + deprecated-pragma, + use-symbolic-message-instead, + apply-builtin, + basestring-builtin, + buffer-builtin, + cmp-builtin, + coerce-builtin, + execfile-builtin, + file-builtin, + long-builtin, + raw_input-builtin, + reduce-builtin, + standarderror-builtin, + unicode-builtin, + xrange-builtin, + coerce-method, + delslice-method, + getslice-method, + setslice-method, + no-absolute-import, + old-division, + dict-iter-method, + dict-view-method, + next-method-called, + metaclass-assignment, + indexing-exception, + raising-string, + reload-builtin, + oct-method, + hex-method, + nonzero-method, + cmp-method, + input-builtin, + round-builtin, + intern-builtin, + unichr-builtin, + map-builtin-not-iterating, + zip-builtin-not-iterating, + range-builtin-not-iterating, + filter-builtin-not-iterating, + using-cmp-argument, + eq-without-hash, + div-method, + idiv-method, + rdiv-method, + exception-message-attribute, + invalid-str-codec, + sys-max-int, + bad-python3-import, + deprecated-string-function, + deprecated-str-translate-call, + deprecated-itertools-function, + deprecated-types-field, + next-method-defined, + dict-items-not-iterating, + dict-keys-not-iterating, + dict-values-not-iterating, + deprecated-operator-function, + deprecated-urllib-function, + xreadlines-attribute, + deprecated-sys-function, + exception-escape, + comprehension-escape + +# Enable the message, report, category or checker with the given id(s). You can +# either give multiple identifier separated by comma (,) or put this option +# multiple time (only on the command line, not in the configuration file where +# it should appear only once). See also the "--disable" option for examples. +enable=c-extension-no-member + + +[REPORTS] + +# Python expression which should return a score less than or equal to 10. You +# have access to the variables 'error', 'warning', 'refactor', and 'convention' +# which contain the number of messages in each category, as well as 'statement' +# which is the total number of statements analyzed. This score is used by the +# global evaluation report (RP0004). +evaluation=10.0 - ((float(5 * error + warning + refactor + convention) / statement) * 10) + +# Template used to display messages. This is a python new-style format string +# used to format the message information. See doc for all details. +#msg-template= + +# Set the output format. Available formats are text, parseable, colorized, json +# and msvs (visual studio). You can also give a reporter class, e.g. +# mypackage.mymodule.MyReporterClass. +output-format=text + +# Tells whether to display a full report or only the messages. +reports=no + +# Activate the evaluation score. +score=yes + + +[REFACTORING] + +# Maximum number of nested blocks for function / method body +max-nested-blocks=5 + +# Complete name of functions that never returns. When checking for +# inconsistent-return-statements if a never returning function is called then +# it will be considered as an explicit return statement and no message will be +# printed. +never-returning-functions=sys.exit + + +[LOGGING] + +# The type of string formatting that logging methods do. `old` means using % +# formatting, `new` is for `{}` formatting. +logging-format-style=old + +# Logging modules to check that the string format arguments are in logging +# function parameter format. +logging-modules=logging + + +[SPELLING] + +# Limits count of emitted suggestions for spelling mistakes. +max-spelling-suggestions=4 + +# Spelling dictionary name. Available dictionaries: none. To make it work, +# install the python-enchant package. +spelling-dict= + +# List of comma separated words that should not be checked. +spelling-ignore-words= + +# A path to a file that contains the private dictionary; one word per line. +spelling-private-dict-file= + +# Tells whether to store unknown words to the private dictionary (see the +# --spelling-private-dict-file option) instead of raising a message. +spelling-store-unknown-words=no + + +[MISCELLANEOUS] + +# List of note tags to take in consideration, separated by a comma. +notes=FIXME, + XXX, + TODO + +# Regular expression of note tags to take in consideration. +#notes-rgx= + + +[TYPECHECK] + +# List of decorators that produce context managers, such as +# contextlib.contextmanager. Add to this list to register other decorators that +# produce valid context managers. +contextmanager-decorators=contextlib.contextmanager + +# List of members which are set dynamically and missed by pylint inference +# system, and so shouldn't trigger E1101 when accessed. Python regular +# expressions are accepted. +generated-members= + +# Tells whether missing members accessed in mixin class should be ignored. A +# mixin class is detected if its name ends with "mixin" (case insensitive). +ignore-mixin-members=yes + +# Tells whether to warn about missing members when the owner of the attribute +# is inferred to be None. +ignore-none=yes + +# This flag controls whether pylint should warn about no-member and similar +# checks whenever an opaque object is returned when inferring. The inference +# can return multiple potential results while evaluating a Python object, but +# some branches might not be evaluated, which results in partial inference. In +# that case, it might be useful to still emit no-member and other checks for +# the rest of the inferred objects. +ignore-on-opaque-inference=yes + +# List of class names for which member attributes should not be checked (useful +# for classes with dynamically set attributes). This supports the use of +# qualified names. +ignored-classes=optparse.Values,thread._local,_thread._local + +# List of module names for which member attributes should not be checked +# (useful for modules/projects where namespaces are manipulated during runtime +# and thus existing member attributes cannot be deduced by static analysis). It +# supports qualified module names, as well as Unix pattern matching. +ignored-modules= + +# Show a hint with possible names when a member name was not found. The aspect +# of finding the hint is based on edit distance. +missing-member-hint=yes + +# The minimum edit distance a name should have in order to be considered a +# similar match for a missing member name. +missing-member-hint-distance=1 + +# The total number of similar names that should be taken in consideration when +# showing a hint for a missing member. +missing-member-max-choices=1 + +# List of decorators that change the signature of a decorated function. +signature-mutators= + + +[VARIABLES] + +# List of additional names supposed to be defined in builtins. Remember that +# you should avoid defining new builtins when possible. +additional-builtins= + +# Tells whether unused global variables should be treated as a violation. +allow-global-unused-variables=yes + +# List of strings which can identify a callback function by name. A callback +# name must start or end with one of those strings. +callbacks=cb_, + _cb + +# A regular expression matching the name of dummy variables (i.e. expected to +# not be used). +dummy-variables-rgx=_+$|(_[a-zA-Z0-9_]*[a-zA-Z0-9]+?$)|dummy|^ignored_|^unused_ + +# Argument names that match this expression will be ignored. Default to name +# with leading underscore. +ignored-argument-names=_.*|^ignored_|^unused_ + +# Tells whether we should check for unused import in __init__ files. +init-import=no + +# List of qualified module names which can have objects that can redefine +# builtins. +redefining-builtins-modules=six.moves,past.builtins,future.builtins,builtins,io + + +[FORMAT] + +# Expected format of line ending, e.g. empty (any line ending), LF or CRLF. +expected-line-ending-format= + +# Regexp for a line that is allowed to be longer than the limit. +ignore-long-lines=^\s*(# )??$ + +# Number of spaces of indent required inside a hanging or continued line. +indent-after-paren=4 + +# String used as indentation unit. This is usually " " (4 spaces) or "\t" (1 +# tab). +indent-string=' ' + +# Maximum number of characters on a single line. +max-line-length=100 + +# Maximum number of lines in a module. +max-module-lines=1000 + +# List of optional constructs for which whitespace checking is disabled. `dict- +# separator` is used to allow tabulation in dicts, etc.: {1 : 1,\n222: 2}. +# `trailing-comma` allows a space between comma and closing bracket: (a, ). +# `empty-line` allows space-only lines. +no-space-check=trailing-comma, + dict-separator + +# Allow the body of a class to be on the same line as the declaration if body +# contains single statement. +single-line-class-stmt=no + +# Allow the body of an if to be on the same line as the test if there is no +# else. +single-line-if-stmt=no + + +[SIMILARITIES] + +# Ignore comments when computing similarities. +ignore-comments=yes + +# Ignore docstrings when computing similarities. +ignore-docstrings=yes + +# Ignore imports when computing similarities. +ignore-imports=no + +# Minimum lines number of a similarity. +min-similarity-lines=4 + + +[BASIC] + +# Naming style matching correct argument names. +argument-naming-style=snake_case + +# Regular expression matching correct argument names. Overrides argument- +# naming-style. +#argument-rgx= + +# Naming style matching correct attribute names. +attr-naming-style=snake_case + +# Regular expression matching correct attribute names. Overrides attr-naming- +# style. +#attr-rgx= + +# Bad variable names which should always be refused, separated by a comma. +bad-names=foo, + bar, + baz, + toto, + tutu, + tata + +# Bad variable names regexes, separated by a comma. If names match any regex, +# they will always be refused +bad-names-rgxs= + +# Naming style matching correct class attribute names. +class-attribute-naming-style=any + +# Regular expression matching correct class attribute names. Overrides class- +# attribute-naming-style. +#class-attribute-rgx= + +# Naming style matching correct class names. +class-naming-style=PascalCase + +# Regular expression matching correct class names. Overrides class-naming- +# style. +#class-rgx= + +# Naming style matching correct constant names. +const-naming-style=UPPER_CASE + +# Regular expression matching correct constant names. Overrides const-naming- +# style. +#const-rgx= + +# Minimum line length for functions/classes that require docstrings, shorter +# ones are exempt. +docstring-min-length=-1 + +# Naming style matching correct function names. +function-naming-style=snake_case + +# Regular expression matching correct function names. Overrides function- +# naming-style. +#function-rgx= + +# Good variable names which should always be accepted, separated by a comma. +good-names=f, + i, + j, + k, + ex, + Run, + _ + +# Good variable names regexes, separated by a comma. If names match any regex, +# they will always be accepted +good-names-rgxs= + +# Include a hint for the correct naming format with invalid-name. +include-naming-hint=no + +# Naming style matching correct inline iteration names. +inlinevar-naming-style=any + +# Regular expression matching correct inline iteration names. Overrides +# inlinevar-naming-style. +#inlinevar-rgx= + +# Naming style matching correct method names. +method-naming-style=snake_case + +# Regular expression matching correct method names. Overrides method-naming- +# style. +#method-rgx= + +# Naming style matching correct module names. +module-naming-style=snake_case + +# Regular expression matching correct module names. Overrides module-naming- +# style. +#module-rgx= + +# Colon-delimited sets of names that determine each other's naming style when +# the name regexes allow several styles. +name-group= + +# Regular expression which should only match function or class names that do +# not require a docstring. +no-docstring-rgx=^_ + +# List of decorators that produce properties, such as abc.abstractproperty. Add +# to this list to register other decorators that produce valid properties. +# These decorators are taken in consideration only for invalid-name. +property-classes=abc.abstractproperty + +# Naming style matching correct variable names. +variable-naming-style=snake_case + +# Regular expression matching correct variable names. Overrides variable- +# naming-style. +#variable-rgx= + + +[STRING] + +# This flag controls whether inconsistent-quotes generates a warning when the +# character used as a quote delimiter is used inconsistently within a module. +check-quote-consistency=no + +# This flag controls whether the implicit-str-concat should generate a warning +# on implicit string concatenation in sequences defined over several lines. +check-str-concat-over-line-jumps=no + + +[IMPORTS] + +# List of modules that can be imported at any level, not just the top level +# one. +allow-any-import-level= + +# Allow wildcard imports from modules that define __all__. +allow-wildcard-with-all=no + +# Analyse import fallback blocks. This can be used to support both Python 2 and +# 3 compatible code, which means that the block might have code that exists +# only in one or another interpreter, leading to false positives when analysed. +analyse-fallback-blocks=no + +# Deprecated modules which should not be used, separated by a comma. +deprecated-modules=optparse,tkinter.tix + +# Create a graph of external dependencies in the given file (report RP0402 must +# not be disabled). +ext-import-graph= + +# Create a graph of every (i.e. internal and external) dependencies in the +# given file (report RP0402 must not be disabled). +import-graph= + +# Create a graph of internal dependencies in the given file (report RP0402 must +# not be disabled). +int-import-graph= + +# Force import order to recognize a module as part of the standard +# compatibility libraries. +known-standard-library= + +# Force import order to recognize a module as part of a third party library. +known-third-party=enchant + +# Couples of modules and preferred modules, separated by a comma. +preferred-modules= + + +[CLASSES] + +# List of method names used to declare (i.e. assign) instance attributes. +defining-attr-methods=__init__, + __new__, + setUp, + __post_init__ + +# List of member names, which should be excluded from the protected access +# warning. +exclude-protected=_asdict, + _fields, + _replace, + _source, + _make + +# List of valid names for the first argument in a class method. +valid-classmethod-first-arg=cls + +# List of valid names for the first argument in a metaclass class method. +valid-metaclass-classmethod-first-arg=cls + + +[DESIGN] + +# Maximum number of arguments for function / method. +max-args=5 + +# Maximum number of attributes for a class (see R0902). +max-attributes=7 + +# Maximum number of boolean expressions in an if statement (see R0916). +max-bool-expr=5 + +# Maximum number of branch for function / method body. +max-branches=12 + +# Maximum number of locals for function / method body. +max-locals=15 + +# Maximum number of parents for a class (see R0901). +max-parents=7 + +# Maximum number of public methods for a class (see R0904). +max-public-methods=20 + +# Maximum number of return / yield for function / method body. +max-returns=6 + +# Maximum number of statements in function / method body. +max-statements=50 + +# Minimum number of public methods for a class (see R0903). +min-public-methods=2 + + +[EXCEPTIONS] + +# Exceptions that will emit a warning when being caught. Defaults to +# "BaseException, Exception". +overgeneral-exceptions=BaseException, + Exception diff --git a/setup.py b/setup.py index 8793fc4..5852b26 100644 --- a/setup.py +++ b/setup.py @@ -92,6 +92,7 @@ def package_vars(version_file): # to never grab the regression version. "coveralls != 1.11.0", "coverage", + "pylint", "pytest-cov", "pytest", ] From 16a158458afcadaf963baabd590bc10672e05b42 Mon Sep 17 00:00:00 2001 From: Nicholas McDonnell <50747025+mcdonnnj@users.noreply.github.com> Date: Tue, 21 Jul 2020 11:54:11 -0400 Subject: [PATCH 2/8] Disable more low-priority tests for pylint --- .pylintrc | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/.pylintrc b/.pylintrc index d56e2cd..dd5c053 100644 --- a/.pylintrc +++ b/.pylintrc @@ -60,7 +60,11 @@ confidence= # --enable=similarities". If you want to run only the classes checker, but have # no Warning level messages displayed, use "--disable=all --enable=classes # --disable=W". -disable=C0330, +disable=too-many-arguments, + too-many-locals, + too-many-return-statements, + line-too-long, + C0330, print-statement, parameter-unpacking, unpacking-in-except, From 46073ffbe9bffcdd3f76173356afeae22521eabe Mon Sep 17 00:00:00 2001 From: Pascal-0x90 Date: Thu, 30 Jul 2020 11:03:20 -0700 Subject: [PATCH 3/8] Fixed pylint errors. Refactored code. Signed-off-by: Pascal-0x90 --- .isort.cfg | 2 +- setup.py | 2 +- src/findcdn/cdnEngine/__init__.py | 6 - src/findcdn/cdnEngine/detectCDN/cdn_check.py | 280 ----------------- src/findcdn/cdnengine/__init__.py | 7 + .../cdnEngine.py => cdnengine/cdnengine.py} | 70 +++-- .../detectcdn}/__init__.py | 8 +- src/findcdn/cdnengine/detectcdn/cdn_check.py | 296 ++++++++++++++++++ .../detectcdn}/cdn_config.py | 0 .../detectcdn}/cdn_err.py | 0 src/findcdn/findcdn.py | 45 +-- src/findcdn/findcdn_err.py | 2 +- tests/conftest.py | 29 -- tests/invalidTest.txt | 17 - tests/test_cdnengine.py | 118 ------- tests/test_detectCDN.py | 182 ----------- tests/test_findcdn.py | 138 -------- tests/validTest.txt | 4 - 18 files changed, 372 insertions(+), 834 deletions(-) delete mode 100644 src/findcdn/cdnEngine/__init__.py delete mode 100755 src/findcdn/cdnEngine/detectCDN/cdn_check.py create mode 100644 src/findcdn/cdnengine/__init__.py rename src/findcdn/{cdnEngine/cdnEngine.py => cdnengine/cdnengine.py} (74%) rename src/findcdn/{cdnEngine/detectCDN => cdnengine/detectcdn}/__init__.py (54%) create mode 100755 src/findcdn/cdnengine/detectcdn/cdn_check.py rename src/findcdn/{cdnEngine/detectCDN => cdnengine/detectcdn}/cdn_config.py (100%) rename src/findcdn/{cdnEngine/detectCDN => cdnengine/detectcdn}/cdn_err.py (100%) delete mode 100644 tests/conftest.py delete mode 100644 tests/invalidTest.txt delete mode 100755 tests/test_cdnengine.py delete mode 100644 tests/test_detectCDN.py delete mode 100644 tests/test_findcdn.py delete mode 100644 tests/validTest.txt diff --git a/.isort.cfg b/.isort.cfg index d7aeec1..2369f81 100644 --- a/.isort.cfg +++ b/.isort.cfg @@ -7,7 +7,7 @@ import_heading_thirdparty=Third-Party Libraries import_heading_firstparty=cisagov Libraries # Should be auto-populated by seed-isort-config hook -known_third_party=dns,docopt,ipwhois,pytest,schema,setuptools,tqdm,validators +known_third_party=dns,docopt,ipwhois,schema,setuptools,tqdm,validators # These must be manually set to correctly separate them from third party libraries known_first_party=findcdn diff --git a/setup.py b/setup.py index 5852b26..b4c4b38 100644 --- a/setup.py +++ b/setup.py @@ -26,7 +26,7 @@ def package_vars(version_file): """Read in and return the variables defined by the version_file.""" pkg_vars = {} with open(version_file) as f: - exec(f.read(), pkg_vars) # nosec + exec(f.read(), pkg_vars) # nosec # pylint: disable=exec-used return pkg_vars diff --git a/src/findcdn/cdnEngine/__init__.py b/src/findcdn/cdnEngine/__init__.py deleted file mode 100644 index ce0487b..0000000 --- a/src/findcdn/cdnEngine/__init__.py +++ /dev/null @@ -1,6 +0,0 @@ -"""cdnEngine library.""" -from . import detectCDN -from .cdnEngine import Chef, DomainPot, run_checks - -"""Define public exports.""" -__all__ = ["DomainPot", "Chef", "run_checks", "detectCDN"] diff --git a/src/findcdn/cdnEngine/detectCDN/cdn_check.py b/src/findcdn/cdnEngine/detectCDN/cdn_check.py deleted file mode 100755 index e68c38c..0000000 --- a/src/findcdn/cdnEngine/detectCDN/cdn_check.py +++ /dev/null @@ -1,280 +0,0 @@ -#!/usr/bin/env python3 - -""" -Summary: This is the main runner for detectCDn. - -Description: The detectCDN library is meant to show what CDNs a domain may be using -""" - -# Standard Python Libraries -from http.client import RemoteDisconnected -from ssl import CertificateError, SSLError -from typing import List -from urllib import request as request -from urllib.error import URLError - -# Third-Party Libraries -from dns.resolver import NXDOMAIN, NoAnswer, NoNameservers, Resolver, Timeout, query -from ipwhois import HTTPLookupError, IPDefinedError, IPWhois -from ipwhois.exceptions import ASNRegistryError - - -# Internal Libraries -from .cdn_config import COMMON, CDNs, CDNs_rev -from .cdn_err import NoIPaddress - -# Global variables -LIFETIME = 10 - - -class Domain: - """Domain class allows for storage of metadata on domain.""" - - def __init__( - self, - url: str, - ip: List[str] = [], - cnames: List[str] = [], - cdns: List[str] = [], - cdns_by_name: List[str] = [], - namsrvs: List[str] = [], - headers: List[str] = [], - whois_data: List[str] = [], - ): - """Initialize object to store metadata on domain in url.""" - self.url = url - self.ip = ip - self.cnames = cnames - self.cdns = cdns - self.cdns_by_name = cdns_by_name - self.namesrvs = namsrvs - self.headers = headers - self.whois_data = whois_data - self.cdn_present = False - - -class cdnCheck: - """cdnCheck runs analysis and stores discovered data in Domain object.""" - - def __init__(self): - """Initialize the orchestrator of analysis.""" - self.running = False - - def ip(self, dom: Domain) -> List[int]: - """Determine IP addresses the domain resolves to.""" - dom_list: List[str] = [dom.url, "www." + dom.url] - return_codes = [] - ip_list = [] - for domain in dom_list: - try: - # Query the domain - response = query(domain) - # Assign any found IP addresses to the object - for ip in response: - if str(ip.address) not in ip_list and str(ip.address) not in dom.ip: - ip_list.append(str(ip.address)) - except NoAnswer: - return_codes.append(1) - except NoNameservers: - return_codes.append(2) - except NXDOMAIN: - return_codes.append(3) - except Timeout: - return_codes.append(4) - - # Append all addresses into IP_list - for addr in ip_list: - dom.ip.append(addr) - # Return listing of error codes - return return_codes - - def cname(self, dom: Domain, timeout: int) -> List[int]: - """Collect CNAME records on domain.""" - # List of domains to check - dom_list = [dom.url, "www." + dom.url] - # Our codes to return - return_code = [] - # Seutp resolver and timeouts - resolver = Resolver() - resolver.timeout = timeout - resolver.lifetime = LIFETIME - cname_query = resolver.query - # Iterate through all domains in list - for domain in dom_list: - try: - response = cname_query(domain, "cname") - dom.cnames = [record.to_text() for record in response] - except NoAnswer: - return_code.append(1) - except NoNameservers: - return_code.append(2) - except NXDOMAIN: - return_code.append(3) - except Timeout: - return_code.append(4) - return return_code - - def https_lookup( - self, dom: Domain, timeout: int, agent: str, interactive: bool, verbose: bool - ) -> int: - """Read 'server' header for CDN hints.""" - # List of domains with different protocols to check. - PROTOCOLS = ["https://", "https://www."] - # Iterate through all protocols - for PROTOCOL in PROTOCOLS: - try: - # Some domains only respond when we have a User-Agent defined. - req = request.Request( - PROTOCOL + dom.url, data=None, headers={"User-Agent": agent}, - ) - # Making the timeout 50 as to not hang thread. - response = request.urlopen(req, timeout=timeout) # nosec - except URLError: - continue - except RemoteDisconnected: - continue - except CertificateError: - continue - except ConnectionResetError: - continue - except SSLError: - continue - except Exception as e: - # Define an exception just in case we missed one. - if interactive or verbose: - print(f"[{e}]: https://{dom.url}") - continue - # Define headers to check for the response - # to grab strings for later parsing. - HEADERS = ["server", "via"] - for value in HEADERS: - if ( - response.headers[value] is not None - and response.headers[value] not in dom.headers - ): - dom.headers.append(response.headers[value]) - return 0 - - def whois(self, dom: Domain, interactive: bool, verbose: bool) -> int: - """Scrape WHOIS data for the org or asn_description.""" - # Make sure we have Ip addresses to check - try: - if len(dom.ip) <= 0: - raise NoIPaddress - except NoIPaddress: - return 1 - # Define temp list to assign - whois_data = [] - # Iterate through all the IP addresses in object - for ip in dom.ip: - try: - response = IPWhois(ip) - # These two should be where we can find substrings hinting to CDN - try: - org = response.lookup_whois()["asn_description"] - if org != "BAREFRUIT-ERRORHANDLING": - whois_data.append(org) - except AttributeError: - pass - try: - org = response.lookup_rdap()["network"]["name"] - if org != "BAREFRUIT-ERRORHANDLING": - whois_data.append(org) - except AttributeError: - pass - except HTTPLookupError: - pass - except IPDefinedError: - pass - except ASNRegistryError: - pass - except Exception as e: - if interactive or verbose: - print(f"[{e}]: {dom.url} for {ip}") - for data in whois_data: - if data not in dom.whois_data: - dom.whois_data.append(data) - # Everything was successful - return 0 - - def CDNid(self, dom: Domain, data_blob: List): - """ - Identify any CDN name in list received. - - All of these will be doing some sort of substring analysis - on each string from any list passed to it. This will help - us identify the CDN which could be used. - """ - for data in data_blob: - # Make sure we do not try to analyze None type data - if data is None: - continue - # Check the CDNs standard list - for url in CDNs: - if ( - url.lower().replace(" ", "") in data.lower().replace(" ", "") - and url not in dom.cdns - ): - dom.cdns.append(url) - dom.cdns_by_name.append(CDNs[url]) - - # Check the CDNs reverse list - for name in CDNs_rev: - if name.lower() in data.lower() and CDNs_rev[name] not in dom.cdns: - dom.cdns.append(CDNs_rev[name]) - dom.cdns_by_name.append(name) - - # Check the CDNs Common list: - for name in COMMON.keys(): - if ( - name.lower().replace(" ", "") in data.lower().replace(" ", "") - and CDNs_rev[name] not in dom.cdns - ): - dom.cdns.append(CDNs_rev[name]) - dom.cdns_by_name.append(name) - - def data_digest(self, dom: Domain) -> int: - """Digest all data collected and assign to CDN list.""" - return_code = 1 - # Iterate through all attributes for substrings - if len(dom.cnames) > 0 and not None: - self.CDNid(dom, dom.cnames) - return_code = 0 - if len(dom.headers) > 0 and not None: - self.CDNid(dom, dom.headers) - return_code = 0 - if len(dom.namesrvs) > 0 and not None: - self.CDNid(dom, dom.namesrvs) - return_code = 0 - if len(dom.whois_data) > 0 and not None: - self.CDNid(dom, dom.whois_data) - return_code = 0 - return return_code - - def all_checks( - self, - dom: Domain, - timeout: int, - agent: str, - verbose: bool = False, - interactive: bool = False, - ) -> int: - """Option to run everything in this library then digest.""" - # Obtain each attributes data - self.ip(dom) - self.cname(dom, timeout) - self.https_lookup(dom, timeout, agent, interactive, verbose) - self.whois(dom, interactive, verbose) - - # Digest the data - return_code = self.data_digest(dom) - - # Extra case if we want verbosity for each domain check - if verbose: - if len(dom.cdns) > 0: - print(f"{dom.url} has the following CDNs:\n{dom.cdns}") - else: - print(f"{dom.url} does not use a CDN") - - # Return to calling function - return return_code diff --git a/src/findcdn/cdnengine/__init__.py b/src/findcdn/cdnengine/__init__.py new file mode 100644 index 0000000..20bb384 --- /dev/null +++ b/src/findcdn/cdnengine/__init__.py @@ -0,0 +1,7 @@ +"""cdnengine library.""" +# Ignoring mypy because this library does infact exist. +from . import detectcdn # type: ignore +from .cdnengine import Chef, DomainPot, run_checks + +# Define public exports +__all__ = ["DomainPot", "Chef", "run_checks", "detectcdn"] diff --git a/src/findcdn/cdnEngine/cdnEngine.py b/src/findcdn/cdnengine/cdnengine.py similarity index 74% rename from src/findcdn/cdnEngine/cdnEngine.py rename to src/findcdn/cdnengine/cdnengine.py index 36519c9..ea4cd56 100755 --- a/src/findcdn/cdnEngine/cdnEngine.py +++ b/src/findcdn/cdnengine/cdnengine.py @@ -1,9 +1,9 @@ #!/usr/bin/env python3 """ -Summary: cdnEngine orchestrates CDN detection of domains. +Summary: cdnengine orchestrates CDN detection of domains. -Description: cdnEngine is a simple solution for detection +Description: cdnengine is a simple solution for detection if a given domain or set of domains use a CDN. """ @@ -18,49 +18,48 @@ # Internal Libraries -from . import detectCDN +from . import detectcdn class DomainPot: """DomainPot defines the "pot" which Domain objects are stored.""" + # pylint: disable=too-few-public-methods + # DomainPot is okay to have only an __init__ as we only need it to have the private var. + def __init__(self, domains: List[str]): """Define the pot for the Chef to use.""" - self.domains: List[detectCDN.Domain] = [] + self.domains: List[detectcdn.Domain] = [] # Convert to list of type domain for dom in domains: - dom_in = detectCDN.Domain( - dom, list(), list(), list(), list(), list(), list(), list() - ) + dom_in = detectcdn.Domain(dom,) self.domains.append(dom_in) def chef_executor( - domain: detectCDN.Domain, + domain: detectcdn.Domain, timeout: int, user_agent: str, verbosity: bool, interactive: bool, ): """Attempt to make the method "threadsafe" by giving each worker its own detector.""" - # Define detector - detective = detectCDN.cdnCheck() - # Run checks try: - detective.all_checks( + detectcdn.all_checks( # Timeout is split by .4 so that each chunk can only take less than half. domain, verbose=verbosity, timeout=math.ceil(timeout * 0.4), agent=user_agent, - interactive=interactive, ) - except Exception as e: + # Allow base exception for when something weird happens. Cannot really find what gets thrown + # since each time it errors its a different unique value. (sometimes based on domain name.) + except BaseException as err: # pylint: disable=broad-except # Incase some uncaught error somewhere if interactive or verbosity: - print(f"An unusual exception has occurred:\n{e}") + print(f"An unusual exception has occurred:\n{err}") return 1 # Return 0 for success @@ -72,15 +71,15 @@ class Chef: def __init__( self, - pot: DomainPot, + pot: List[str], threads: int, timeout: int, user_agent: str, interactive: bool = False, verbose: bool = False, ): - """Give the chef the pot to use.""" - self.pot: DomainPot = pot + """Give the chef the domain pot to use.""" + self.pot: List[detectcdn.Domain] = list() self.pbar: tqdm = interactive self.verbose: bool = verbose self.timeout: int = timeout @@ -98,6 +97,11 @@ def __init__( cpu_count = 1 self.threads = cpu_count # type: ignore + # Convert pot to domain objects + for dom in pot: + dom_in = detectcdn.Domain(dom,) + self.pot.append(dom_in) + def grab_cdn( self, double: bool = False # type: ignore ): @@ -116,9 +120,9 @@ def grab_cdn( # If double, Double contents to combat CDN cache misses newpot = [] if double: - for domain in self.pot.domains: + for domain in self.pot: newpot.append(domain) - for domain in self.pot.domains: + for domain in self.pot: newpot.append(domain) job_count = len(newpot) # Setup pbar with correct amount size @@ -138,24 +142,27 @@ def grab_cdn( for domain in newpot } + # Init completed ammount + completed: int = 0 # Comb future objects for completed task pool. for future in concurrent.futures.as_completed(results): try: # Try and grab feature result to dequeue job future.result(timeout=self.timeout) - except concurrent.futures.TimeoutError as e: + except concurrent.futures.TimeoutError as err: # Tell us we dropped it. Should log this instead. if self.interactive or self.verbose: - print(f"Dropped due to: {e}") + print(f"Dropped due to: {err}") # Update status bar if allowed if self.pbar: - # We type ignore these as its "illegal" to access private attributes of an object - pending = f"Pending: {executor._work_queue.qsize()} jobs" # type: ignore - threads = f"Threads: {len(executor._threads)}" # type: ignore + # We type ignore these as its "illegal" to access private attributes of an object but we need to (thanks python) + pending = f"Pending: {job_count - completed} jobs" + threads = f"Threads: {self.threads}" pbar.set_description(f"[{pending}]==[{threads}]") if self.pbar is not None: pbar.update(1) + completed += 1 else: pass @@ -164,12 +171,12 @@ def grab_cdn( def has_cdn(self): """For each domain, check if domain contains CDNS. If so, tick cdn_present to true.""" - for domain in self.pot.domains: + for domain in self.pot: if len(domain.cdns) > 0: domain.cdn_present = True def run_checks(self, double: bool = False) -> int: - """Run analysis on the internal domain pool using detectCDN library.""" + """Run analysis on the internal domain pool using detectcdn library.""" cnt = self.grab_cdn(double) self.has_cdn() return cnt @@ -183,16 +190,13 @@ def run_checks( interactive: bool = False, verbose: bool = False, double: bool = False, -) -> Tuple[List[detectCDN.Domain], int]: +) -> Tuple[List[detectcdn.Domain], int]: """Orchestrate the use of DomainPot and Chef.""" - # Our domain pot - dp = DomainPot(domains) - # Our chef to manage pot - chef = Chef(dp, threads, timeout, user_agent, interactive, verbose) + chef = Chef(domains, threads, timeout, user_agent, interactive, verbose) # Run analysis for all domains cnt = chef.run_checks(double) # Return all domains in form domain_pool, count of jobs processed, error code - return (chef.pot.domains, cnt) + return (chef.pot, cnt) diff --git a/src/findcdn/cdnEngine/detectCDN/__init__.py b/src/findcdn/cdnengine/detectcdn/__init__.py similarity index 54% rename from src/findcdn/cdnEngine/detectCDN/__init__.py rename to src/findcdn/cdnengine/detectcdn/__init__.py index 17b39d4..82098f4 100644 --- a/src/findcdn/cdnEngine/detectCDN/__init__.py +++ b/src/findcdn/cdnengine/detectcdn/__init__.py @@ -1,11 +1,15 @@ """detectCDN Library.""" -from .cdn_check import Domain, cdnCheck +from .cdn_check import Domain, all_checks, cname, https_lookup, ip_resolve, whois from .cdn_config import COMMON, CDNs, CDNs_rev from .cdn_err import NoIPaddress __all__ = [ "Domain", - "cdnCheck", + "all_checks", + "ip_resolve", + "cname", + "https_lookup", + "whois", "CDNs_rev", "CDNs", "COMMON", diff --git a/src/findcdn/cdnengine/detectcdn/cdn_check.py b/src/findcdn/cdnengine/detectcdn/cdn_check.py new file mode 100755 index 0000000..e413749 --- /dev/null +++ b/src/findcdn/cdnengine/detectcdn/cdn_check.py @@ -0,0 +1,296 @@ +#!/usr/bin/env python3 + +""" +Summary: This is the main runner for detectCDn. + +Description: The detectCDN library is meant to show what CDNs a domain may be using +""" + +# Standard Python Libraries +from http.client import HTTPException, RemoteDisconnected +import socket +from ssl import CertificateError, SSLError +from typing import List +from urllib import request +from urllib.error import HTTPError, URLError + +# Third-Party Libraries +from dns.resolver import NXDOMAIN, NoAnswer, NoNameservers, Resolver, Timeout, query +from ipwhois import HTTPLookupError, IPDefinedError, IPWhois +from ipwhois.exceptions import ASNRegistryError + + +# Internal Libraries +from .cdn_config import COMMON, CDNs, CDNs_rev +from .cdn_err import NoIPaddress + +# Global variables +LIFETIME = 10 + + +class Domain: + """Domain class allows for storage of metadata on domain.""" + + # pylint: disable=too-many-instance-attributes + # 9 is the amount of attributes we need this object to have + # pylint: disable=too-few-public-methods + # This class doesnt need methods since we can call . on the object. + + def __init__( + self, url: str, + ): + """Initialize object to store metadata on domain in url.""" + self.url = url + self.ip_list: List[str] = list() + self.cnames: List[str] = list() + self.cdns: List[str] = list() + self.cdns_by_name: List[str] = list() + self.headers: List[str] = list() + self.whois_data: List[str] = list() + + +def ip_resolve(dom: Domain) -> List[int]: + """Determine IP addresses the domain resolves to.""" + # Check if we have initialized the ip list + if dom.ip_list is None: + dom.ip_list = list() + # Define our list of variant urls + dom_list: List[str] = [dom.url, "www." + dom.url] + # Lists to contain data from runs + return_codes = [] + ip_list = [] + # Iterate over domains for the ip addresses + for domain in dom_list: + try: + # Query the domain + response = query(domain) + # Assign any found IP addresses to the object + for ip_iter in response: + if ( + str(ip_iter.address) not in ip_list + and str(ip_iter.address) not in dom.ip_list + ): + ip_list.append(str(ip_iter.address)) + except NoAnswer: + return_codes.append(1) + except NoNameservers: + return_codes.append(2) + except NXDOMAIN: + return_codes.append(3) + except Timeout: + return_codes.append(4) + except (socket.timeout, URLError, HTTPError): + pass + + # Append all addresses into IP_list + for addr in ip_list: + dom.ip_list.append(addr) + # Return listing of error codes + return return_codes + + +def cname(dom: Domain, timeout: int) -> List[int]: + """Collect CNAME records on domain.""" + # Make sure we initialized the cname list + if dom.cnames is None: + dom.cnames = list() + # List of domains to check + dom_list = [dom.url, "www." + dom.url] + # Our codes to return + return_code = [] + # Seutp resolver and timeouts + resolver = Resolver() + resolver.timeout = timeout + resolver.lifetime = LIFETIME + cname_query = resolver.query + # Iterate through all domains in list + for domain in dom_list: + try: + response = cname_query(domain, "cname") + dom.cnames = [record.to_text() for record in response] + except NoAnswer: + return_code.append(1) + except NoNameservers: + return_code.append(2) + except NXDOMAIN: + return_code.append(3) + except Timeout: + return_code.append(4) + except (socket.timeout, URLError, HTTPError): + pass + return return_code + + +def https_lookup(dom: Domain, timeout: int, agent: str) -> int: + """Read 'server' header for CDN hints.""" + # Check if we have initialized the headers list + if dom.headers is None: + dom.headers = list() + # List of domains with different protocols to check. + protocols = ["https://", "https://www."] + # Iterate through all protocols + for protocol in protocols: + try: + # Some domains only respond when we have a User-Agent defined. + req = request.Request( + protocol + dom.url, data=None, headers={"User-Agent": agent}, + ) + # Making the timeout 50 as to not hang thread. + response = request.urlopen(req, timeout=timeout) # nosec + except ( + HTTPError, + URLError, + RemoteDisconnected, + CertificateError, + ConnectionResetError, + SSLError, + Timeout, + socket.timeout, + socket.error, + HTTPException, + ): + continue + # Define headers to check for the response + # to grab strings for later parsing. + headers = ["server", "via"] + for value in headers: + if ( + response.headers[value] is not None + and response.headers[value] not in dom.headers + ): + dom.headers.append(response.headers[value]) + return 0 + + +def whois(dom: Domain) -> int: + """Scrape WHOIS data for the org or asn_description.""" + # pylint: disable=too-many-branches + + # Unfortunately, need all of these branches to catch specific + # errors as not all servers respond the same + + # Make sure we have Ip addresses to check + try: + if dom.ip_list is None or len(dom.ip_list) <= 0: + raise NoIPaddress + except NoIPaddress: + return 1 + # Make sure our domain object has initialized whois data list + if dom.whois_data is None: + dom.whois_data = list() + # Define temp list to assign + whois_data = [] + # Iterate through all the IP addresses in object + for ip_val in dom.ip_list: + try: + response = IPWhois(ip_val) + # These two should be where we can find substrings hinting to CDN + try: + org = response.lookup_whois()["asn_description"] + if org != "BAREFRUIT-ERRORHANDLING": + whois_data.append(org) + except AttributeError: + pass + try: + org = response.lookup_rdap()["network"]["name"] + if org != "BAREFRUIT-ERRORHANDLING": + whois_data.append(org) + except AttributeError: + pass + except ( + HTTPLookupError, + HTTPError, + URLError, + IPDefinedError, + ASNRegistryError, + Timeout, + socket.timeout, + ): + pass + + for data in whois_data: + if data not in dom.whois_data: + dom.whois_data.append(data) + # Everything was successful + return 0 + + +def cdn_id(dom: Domain, data_blob: List): + """ + Identify any CDN name in list received. + + All of these will be doing some sort of substring analysis + on each string from any list passed to it. This will help + us identify the CDN which could be used. + """ + # Check if our cdn list is initialized + if dom.cdns_by_name is None or dom.cdns is None: + dom.cdns = list() + dom.cdns_by_name = list() + + # Iterate over all the data sent in + for data in data_blob: + # Make sure we do not try to analyze None type data + if data is None: + continue + # Check the CDNs standard list + for url in CDNs: + if ( + url.lower().replace(" ", "") in data.lower().replace(" ", "") + and url not in dom.cdns + ): + dom.cdns.append(url) + dom.cdns_by_name.append(CDNs[url]) + + # Check the CDNs reverse list + for name in CDNs_rev: + if name.lower() in data.lower() and CDNs_rev[name] not in dom.cdns: + dom.cdns.append(CDNs_rev[name]) + dom.cdns_by_name.append(name) + + # Check the CDNs Common list: + for name in COMMON: + if ( + name.lower().replace(" ", "") in data.lower().replace(" ", "") + and CDNs_rev[name] not in dom.cdns + ): + dom.cdns.append(CDNs_rev[name]) + dom.cdns_by_name.append(name) + + +def data_digest(dom: Domain) -> int: + """Digest all data collected and assign to CDN list.""" + return_code = 1 + # Iterate through all attributes for substrings + if len(dom.cnames) > 0 and not None: + cdn_id(dom, dom.cnames) + return_code = 0 + if len(dom.headers) > 0 and not None: + cdn_id(dom, dom.headers) + return_code = 0 + if len(dom.whois_data) > 0 and not None: + cdn_id(dom, dom.whois_data) + return_code = 0 + return return_code + + +def all_checks(dom: Domain, timeout: int, agent: str, verbose: bool = False,) -> int: + """Option to run everything in this library then digest.""" + # Obtain each attributes data + ip_resolve(dom) + cname(dom, timeout) + https_lookup(dom, timeout, agent) + whois(dom) + + # Digest the data + return_code = data_digest(dom) + + # Extra case if we want verbosity for each domain check + if verbose: + if len(dom.cdns) > 0: + print(f"{dom.url} has the following CDNs:\n{dom.cdns}") + else: + print(f"{dom.url} does not use a CDN") + + # Return to calling function + return return_code diff --git a/src/findcdn/cdnEngine/detectCDN/cdn_config.py b/src/findcdn/cdnengine/detectcdn/cdn_config.py similarity index 100% rename from src/findcdn/cdnEngine/detectCDN/cdn_config.py rename to src/findcdn/cdnengine/detectcdn/cdn_config.py diff --git a/src/findcdn/cdnEngine/detectCDN/cdn_err.py b/src/findcdn/cdnengine/detectcdn/cdn_err.py similarity index 100% rename from src/findcdn/cdnEngine/detectCDN/cdn_err.py rename to src/findcdn/cdnengine/detectcdn/cdn_err.py diff --git a/src/findcdn/findcdn.py b/src/findcdn/findcdn.py index 14ab714..df50205 100755 --- a/src/findcdn/findcdn.py +++ b/src/findcdn/findcdn.py @@ -45,24 +45,24 @@ # Internal Libraries from ._version import __version__ -from .cdnEngine import run_checks +from .cdnengine import run_checks from .findcdn_err import FileWriteError, InvalidDomain, NoDomains, OutputFileExists # Global Variables USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/35.0.1916.47 Safari/537.36" TIMEOUT = 60 # Time in seconds -THREADS = 0 # If 0 then cdnEngine uses CPU count to set thread count +THREADS = 0 # If 0 then cdnengine uses CPU count to set thread count -def write_json(json_dump: str, output: str, verbose: bool, interactive: bool): +def write_json(json_dump: str, output: str): """Write dict as JSON to output file.""" try: with open(output, "x") as outfile: outfile.write(json_dump) except FileExistsError: raise OutputFileExists(output) - except Exception as e: - raise FileWriteError(e) + except PermissionError as err: + raise FileWriteError(err) def main( @@ -70,7 +70,7 @@ def main( output_path: str = None, verbose: bool = False, all_domains: bool = False, - interactive: bool = False, + interactive_var: bool = False, double_in: bool = False, threads: int = THREADS, timeout: int = TIMEOUT, @@ -79,7 +79,7 @@ def main( """Take in a list of domains and determine the CDN for each return (JSON, number of successful jobs).""" # Make sure the list passed is got something in it if len(domain_list) <= 0: - raise NoDomains("error") + raise NoDomains() # Validate domains in list for item in domain_list: @@ -92,23 +92,23 @@ def main( # Define domain dict and counter for json domain_dict = {} - CDN_count = 0 + cdn_count = 0 # Check domain list processed_list, cnt = run_checks( - domain_list, threads, timeout, user_agent, interactive, verbose, double_in, + domain_list, threads, timeout, user_agent, interactive_var, verbose, double_in, ) # Parse the domain data for domain in processed_list: # Track the count of the domain has cdns if len(domain.cdns) > 0: - CDN_count += 1 + cdn_count += 1 # Setup formatting for json output if len(domain.cdns) > 0 or all_domains: domain_dict[domain.url] = { - "IP": str(domain.ip)[1:-1], + "IP": str(domain.ip_list)[1:-1], "cdns": str(domain.cdns)[1:-1], "cdns_by_names": str(domain.cdns_by_name)[1:-1], } @@ -116,21 +116,21 @@ def main( # Create JSON from the results and return (results, successful jobs) json_dict = {} json_dict["date"] = datetime.datetime.now().strftime("%m/%d/%Y, %H:%M:%S") - json_dict["cdn_count"] = str(CDN_count) + json_dict["cdn_count"] = str(cdn_count) json_dict["domains"] = domain_dict # type: ignore json_dump = json.dumps(json_dict, indent=4, sort_keys=False) # Show the dump to stdout if verbose or interactive - if (output_path is None and interactive) or verbose: + if (output_path is None and interactive_var) or verbose: print(json_dump) # Export to file if file provided if output_path is not None: - write_json(json_dump, output_path, verbose, interactive) - if interactive or verbose: + write_json(json_dump, output_path) + if interactive_var or verbose: print( "Domain processing completed.\n%d domains had CDN's out of %d." - % (CDN_count, len(domain_list)) + % (cdn_count, len(domain_list)) ) if verbose: print(f"{cnt} jobs completed!") @@ -167,7 +167,9 @@ def interactive() -> int: None, And( str, - lambda filename: os.path.isfile(filename), + lambda filename: os.path.isfile( # pylint: disable=unnecessary-lambda + filename + ), error='Input file "' + str(args[""]) + '" does not exist!', ), ), @@ -199,8 +201,8 @@ def interactive() -> int: try: with open(validated_args[""]) as f: domain_list = [line.rstrip() for line in f] - except IOError as e: - print("A file error occurred: %s" % e, file=sys.stderr) + except IOError as err: + print("A file error occurred: %s" % err, file=sys.stderr) return 1 else: domain_list = validated_args[""] @@ -228,12 +230,11 @@ def interactive() -> int: except InvalidDomain as invdom: print(invdom.message) return 3 - except NoDomains as nd: - print(nd.message) + except NoDomains as nde: + print(nde.message) return 4 return 0 if __name__ == "__main__": - """Launch program in interactive mode""" sys.exit(interactive()) diff --git a/src/findcdn/findcdn_err.py b/src/findcdn/findcdn_err.py index 7913141..b8022df 100644 --- a/src/findcdn/findcdn_err.py +++ b/src/findcdn/findcdn_err.py @@ -7,7 +7,7 @@ class NoDomains(Exception): """Raise when no domains are passed to findcdn main.""" - def __init__(self, error): + def __init__(self): """Instantiate super class with passed message.""" self.message = "No domains were passed!" super().__init__(self.message) diff --git a/tests/conftest.py b/tests/conftest.py deleted file mode 100644 index 39e2be2..0000000 --- a/tests/conftest.py +++ /dev/null @@ -1,29 +0,0 @@ -"""pytest plugin configuration. - -https://docs.pytest.org/en/latest/writing_plugins.html#conftest-py-plugins -""" -# Third-Party Libraries -import pytest - - -def pytest_addoption(parser): - """Add new commandline options to pytest.""" - parser.addoption( - "--runslow", action="store_true", default=False, help="run slow tests" - ) - - -def pytest_configure(config): - """Register new markers.""" - config.addinivalue_line("markers", "slow: mark test as slow") - - -def pytest_collection_modifyitems(config, items): - """Modify collected tests based on custom marks and commandline options.""" - if config.getoption("--runslow"): - # --runslow given in cli: do not skip slow tests - return - skip_slow = pytest.mark.skip(reason="need --runslow option to run") - for item in items: - if "slow" in item.keywords: - item.add_marker(skip_slow) diff --git a/tests/invalidTest.txt b/tests/invalidTest.txt deleted file mode 100644 index 6986e0f..0000000 --- a/tests/invalidTest.txt +++ /dev/null @@ -1,17 +0,0 @@ -notadomain -login.gov -duo.com -www.secureworks.com -status.symantec.com -dnc.org -fitbit.com -zillow.com -ethereum.org -refer.walmart.com -www.netgear.com -view.yahoo.com -today.msnbc.com -www.amazon.co.uk -google.com -facebook.com -asdf.com diff --git a/tests/test_cdnengine.py b/tests/test_cdnengine.py deleted file mode 100755 index 95680cc..0000000 --- a/tests/test_cdnengine.py +++ /dev/null @@ -1,118 +0,0 @@ -#!/usr/bin/env pytest -vs -"""Tests for cdnEngine.""" - -# cisagov Libraries -import findcdn - -# Test Global Variables -USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/35.0.1916.47 Safari/537.36" -TIMEOUT = 30 -THREADS = 0 # If 0 then cdnEngine uses CPU count to set thread count - - -def test_domainpot_init(): - """Test if DomainPot can be instantiated correctly.""" - domains = ["asu.edu", "login.gov", "censys.io", "realpython.com"] - pot = findcdn.cdnEngine.DomainPot(domains) - - # Assertions - for i in range(len(domains)): - assert pot.domains[i].url == domains[i], "{} is not {}".format( - pot.domains[i].url, domains[i] - ) - for j in range(len(domains)): - if j == i: - continue - assert id(pot.domains[i].cdns) != id( - pot.domains[j].cdns - ), "Domain obect {} shares address with Domain object {}".format(i, j) - - -def test_chef_init(): - """Test if Chef can be instantiated correctly.""" - domains = ["asu.edu", "login.gov", "censys.io", "realpython.com"] - pot = findcdn.cdnEngine.DomainPot(domains) - chef = findcdn.cdnEngine.Chef(pot, THREADS, TIMEOUT, USER_AGENT) - - # Assertions - assert type(chef.pot) == findcdn.cdnEngine.DomainPot - - -def test_grab_cdn(): - """Test if Chef can obtain proper CDNs of domains.""" - domains = ["asu.edu", "login.gov", "censys.io", "realpython.com"] - pot = findcdn.cdnEngine.DomainPot(domains) - chef = findcdn.cdnEngine.Chef(pot, THREADS, TIMEOUT, USER_AGENT) - chef.run_checks() - checked_domains = chef.pot.domains - - # Assertions - assert checked_domains[0].cdns == [ - ".cloudflare.net", - ".cloudflare.com", - ], "Did not detect {} from {}.".format( - [".cloudflare.net", ".cloudflare.com"], checked_domains[0].url - ) - assert checked_domains[1].cdns == [ - ".cloudfront.net", - ], "Did not detect {} from {}.".format([".cloudfront.net"], checked_domains[1].url) - assert checked_domains[2].cdns == [ - ".cloudflare.com" - ], "Did not detect {} from {}.".format([".cloudflare.com"], checked_domains[2].url) - assert checked_domains[3].cdns == [ - ".cloudflare.com" - ], "Did not detect {} from {}.".format([".cloudflare.com"], checked_domains[3].url) - - -def test_has_cdn(): - """Test that of a set of domains with a without a CDN return correctly.""" - domains = ["asu.edu", "censys.io", "bannerhealth.com", "adobe.com"] - pot = findcdn.cdnEngine.DomainPot(domains) - chef = findcdn.cdnEngine.Chef(pot, THREADS, TIMEOUT, USER_AGENT) - chef.run_checks() - checked_domains = chef.pot.domains - - # Assertions - cdn_present = 0 - for dom in checked_domains: - if dom.cdn_present: - cdn_present += 1 - - assert cdn_present == 3, "Too many cdn_present domains counted." - assert checked_domains[0].url == "asu.edu" and checked_domains[0].cdns == [ - ".cloudflare.net", - ".cloudflare.com", - ], ("Incorrect CDN detected for %s" % checked_domains[0].url) - assert checked_domains[1].url == "censys.io" and checked_domains[1].cdns == [ - ".cloudflare.com" - ], ("Incorrect CDN detected for %s" % checked_domains[1].url) - - -def test_run_checks(): - """Test the run_checks orchestator works.""" - domains = ["asu.edu", "censys.io", "bannerhealth.com", "adobe.com"] - pot = findcdn.cdnEngine.DomainPot(domains) - chef = findcdn.cdnEngine.Chef(pot, THREADS, TIMEOUT, USER_AGENT) - chef.run_checks() - - # Assertions - assert len(chef.pot.domains) > 0, "Pot not stored correctly." - - -def test_run_checks_present(): - """Test the return of a list of cdn_present domains.""" - domains = ["asu.edu", "censys.io", "bannerhealth.com", "adobe.com"] - objects, cnt = findcdn.cdnEngine.run_checks(domains, THREADS, TIMEOUT, USER_AGENT) - cdn_present = {} - for dom in objects: - if dom.cdn_present: - cdn_present[dom.url] = dom.cdns - expected = { - "asu.edu": [".cloudflare.net", ".cloudflare.com"], - "censys.io": [".cloudflare.com"], - "adobe.com": [".edgekey.net", ".akamaitechnologies.fr"], - } - - # Assertions - assert len(cdn_present) > 0, "Returned cdn_present list is empty." - assert cdn_present == expected, "Returned domains do not match." diff --git a/tests/test_detectCDN.py b/tests/test_detectCDN.py deleted file mode 100644 index f2c90b4..0000000 --- a/tests/test_detectCDN.py +++ /dev/null @@ -1,182 +0,0 @@ -#!/usr/bin/env pytest -vs -"""Tests for detectCDN.""" - -# Third-Party Libraries -import dns.resolver - -# cisagov Libraries -from findcdn.cdnEngine.detectCDN import Domain, cdnCheck - -# Globals -TIMEOUT = 10 -USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/35.0.1916.47 Safari/537.36" - - -def test_ip(): - """Test the IP resolving feature.""" - dns.resolver.default_resolver = dns.resolver.Resolver() - dns.resolver.default_resolver.nameservers = ["1.1.1.1", "8.8.8.8"] - dom_in = Domain( - "dns.google.com", list(), list(), list(), list(), list(), list(), list() - ) - check = cdnCheck() - check.ip(dom_in) - - assert "8.8.8.8" in dom_in.ip, "the ip for dns.google.com should be 8.8.8.8" - - -def test_broken_ip(): - """Test a non-working domain IP resolving feature.""" - dns.resolver.default_resolver = dns.resolver.Resolver() - dns.resolver.default_resolver.nameservers = ["1.1.1.1", "8.8.8.8"] - dom_in = Domain( - "notarealdomain.fakedomaindne.com", - list(), - list(), - list(), - list(), - list(), - list(), - list(), - ) - check = cdnCheck() - return_code = check.ip(dom_in) - assert return_code != 0, "This fake site should return a non 0 code." - - -def test_cname(): - """Test the CNAME resolving feature.""" - dom_in = Domain( - "www.asu.edu", list(), list(), list(), list(), list(), list(), list() - ) - check = cdnCheck() - check.cname(dom_in, timeout=TIMEOUT) - - assert ( - "www.asu.edu.cdn.cloudflare.net." in dom_in.cnames - ), "www.asu.edu should have www.asu.edu.cdn.cloudflare.net. as a cname" - - -def test_broken_cname(): - """Test a non-working domain CNAME resolving feature.""" - dom_in = Domain( - "notarealdomain.fakedomaindne.com", - list(), - list(), - list(), - list(), - list(), - list(), - list(), - ) - check = cdnCheck() - return_code = check.cname(dom_in, timeout=TIMEOUT) - assert return_code != 0, "This fake site should return a non 0 code." - - -def test_https_lookup(): - """Test the header resolving feature.""" - dom_in = Domain( - "google.com", list(), list(), list(), list(), list(), list(), list() - ) - check = cdnCheck() - check.https_lookup( - dom_in, timeout=TIMEOUT, agent=USER_AGENT, interactive=False, verbose=False - ) - - assert "gws" in dom_in.headers, "google.com should have gws as a header" - - -def test_broken_https_lookup(): - """Test a non-working domain header resolving feature.""" - dom_in = Domain( - "notarealdomain.fakedomaindne.com", - list(), - list(), - list(), - list(), - list(), - list(), - list(), - ) - check = cdnCheck() - check.https_lookup( - dom_in, timeout=TIMEOUT, agent=USER_AGENT, interactive=False, verbose=False - ) - assert len(dom_in.headers) <= 0, "There should be no response." - - -def test_whois(): - """Test the whois resolving feature.""" - dom_in = Domain( - "google.com", list(), list(), list(), list(), list(), list(), list() - ) - check = cdnCheck() - check.ip(dom_in) - check.whois(dom_in, interactive=False, verbose=False) - - assert ( - "GOOGLE" in dom_in.whois_data - ), "google.com should return GOOGLE in the whois_data" - - -def test_broken_whois(): - """Test a non-working domain whois resolving feature.""" - dom_in = Domain( - "notarealdomain.fakedomaindne.com", - list(), - list(), - list(), - list(), - list(), - list(), - list(), - ) - check = cdnCheck() - check.ip(dom_in) - return_code = check.whois(dom_in, interactive=False, verbose=False) - assert return_code != 0, "This fake site should return a non 0 code." - - -def test_all_checks(): - """Run all checks.""" - dom_in = Domain("login.gov", list(), list(), list(), list(), list(), list(), list()) - check = cdnCheck() - check.all_checks(dom_in, timeout=TIMEOUT, agent=USER_AGENT) - - assert ( - ".cloudfront.net" in dom_in.cdns - ), "the ip for dns.google.com should be 8.8.8.8" - - -def test_all_checks_by_name(): - """Run all checks and get CDN name.""" - dom_in = Domain("login.gov", list(), list(), list(), list(), list(), list(), list()) - check = cdnCheck() - check.all_checks(dom_in, timeout=TIMEOUT, agent=USER_AGENT) - - assert ( - ".cloudfront.net" in dom_in.cdns - ), "the ip for dns.google.com should be 8.8.8.8" - - -def test_all_checks_bad(): - """Test fake domain and ensure it dosen't break anything.""" - dns.resolver.default_resolver = dns.resolver.Resolver() - dns.resolver.default_resolver.nameservers = ["1.1.1.1", "8.8.8.8"] - dom = Domain( - "super.definitelynot.notarealdomain.fakedomaindne.com", - list(), - list(), - list(), - list(), - list(), - list(), - list(), - ) - print(dom.url, dom.cdns, dom.cnames, dom.headers, dom.whois_data, dom.ip) - check = cdnCheck() - return_code = check.all_checks(dom, timeout=TIMEOUT, agent=USER_AGENT) - print(return_code) - print(dom.url, dom.cdns, dom.cnames, dom.headers, dom.whois_data, dom.ip) - assert return_code != 0, "This fake site should return a non 0 code." diff --git a/tests/test_findcdn.py b/tests/test_findcdn.py deleted file mode 100644 index ed2a5b9..0000000 --- a/tests/test_findcdn.py +++ /dev/null @@ -1,138 +0,0 @@ -#!/usr/bin/env pytest -vs -"""Tests for findcdn.""" - -# Standard Python Libraries -import os -import sys -from unittest.mock import patch - -# Third-Party Libraries -import pytest - -# cisagov Libraries -import findcdn - -# define sources of version strings -RELEASE_TAG = os.getenv("RELEASE_TAG") -PROJECT_VERSION = findcdn.__version__ - - -def test_stdout_version(capsys): - """Verify that version string sent to stdout agrees with the module version.""" - with pytest.raises(SystemExit): - with patch.object(sys, "argv", ["bogus", "--version"]): - findcdn.findcdn.interactive() - captured = capsys.readouterr() - assert ( - captured.out == f"{PROJECT_VERSION}\n" - ), "standard output by '--version' should agree with module.__version__" - - -@pytest.mark.skipif( - RELEASE_TAG in [None, ""], reason="this is not a release (RELEASE_TAG not set)" -) -def test_release_version(): - """Verify that release tag version agrees with the module version.""" - assert ( - RELEASE_TAG == f"v{PROJECT_VERSION}" - ), "RELEASE_TAG does not match the project version" - - -def test_list_working(): - """Working domain list to test with.""" - with patch.object( - sys, "argv", ["bogus", "list", "google.com", "facebook.com", "login.gov"] - ): - return_code = findcdn.findcdn.interactive() - assert return_code == 0, "interactive() should return successfully" - - -def test_list_working_double(capsys): - """Working domain list to test -d with.""" - with patch.object( - sys, - "argv", - ["bogus", "list", "google.com", "facebook.com", "login.gov", "-v", "-d"], - ): - return_code = findcdn.findcdn.interactive() - captured = capsys.readouterr() - assert return_code == 0, "interactive() should return successfully" - assert "6 jobs completed" in captured.out - - -def test_list_working_verbose(capsys): - """Working domain list to test -v with.""" - with patch.object( - sys, - "argv", - [ - "bogus", - "list", - "google.com", - "facebook.com", - "superfake.thisisnotarealwebsiteforsure.com", - "-v", - ], - ): - return_code = findcdn.findcdn.interactive() - captured = capsys.readouterr() - assert return_code == 0, "interactive() should return successfully" - assert "3 Domains Validated" in captured.out - - -def test_list_working_tcount(capsys): - """Working domain list to test -t with.""" - with patch.object( - sys, - "argv", - ["bogus", "list", "google.com", "facebook.com", "login.gov", "-t", "3", "-v"], - ): - return_code = findcdn.findcdn.interactive() - captured = capsys.readouterr() - assert return_code == 0, "interactive() should return successfully" - assert "Using 3 threads" in captured.out - - -def test_list_broken(): - """Broken domain list to test with.""" - with patch.object( - sys, - "argv", - ["bogus", "list", "google.com/searchtest", "facebook.com", "login.gov"], - ): - return_code = findcdn.findcdn.interactive() - assert return_code == 3, "interactive() should return failure" - - -def test_file_working(): - """Working domain file to test with.""" - with patch.object(sys, "argv", ["./findcdn", "file", "tests/validTest.txt"]): - return_code = findcdn.findcdn.interactive() - assert return_code == 0, "interactive() should return successfully" - - -"""Test a broken domains passed in as a file""" - - -def test_file_broken(): - """Broken domain file to test with.""" - with patch.object(sys, "argv", ["bogus", "file", "tests/invalidTest.txt"]): - return_code = findcdn.findcdn.interactive() - assert return_code != 0, "interactive() should return failure" - - -def test_file_dne(): - """Working domain list to test with.""" - with patch.object(sys, "argv", ["./findcdn", "file", "nosuchfile.txt"]): - return_code = findcdn.findcdn.interactive() - assert return_code != 0, "interactive() should return successfully" - - -def test_file_write(tmpdir): - """Test writing to a file.""" - file = tmpdir.join("outputtest.txt") - with patch.object( - sys, "argv", ["./findcdn", "list", "google.com", "-o", str(file)] - ): - findcdn.findcdn.interactive() - assert "google.com" in file.read() diff --git a/tests/validTest.txt b/tests/validTest.txt deleted file mode 100644 index 7288725..0000000 --- a/tests/validTest.txt +++ /dev/null @@ -1,4 +0,0 @@ -login.gov -duo.com -facebook.com -asdf.com From 6dce6de16c32b4f066997a518e4233e8f13e93a1 Mon Sep 17 00:00:00 2001 From: Pascal-0x90 Date: Fri, 31 Jul 2020 07:09:01 -0700 Subject: [PATCH 4/8] Readded fixed tests. Signed-off-by: Pascal-0x90 --- .isort.cfg | 2 +- src/findcdn/cdnengine/__init__.py | 4 +- src/findcdn/cdnengine/cdnengine.py | 16 ---- tests/conftest.py | 29 +++++++ tests/invalidTest.txt | 17 ++++ tests/test_cdnengine.py | 98 +++++++++++++++++++++ tests/test_detectcdn.py | 121 ++++++++++++++++++++++++++ tests/test_findcdn.py | 135 +++++++++++++++++++++++++++++ tests/validTest.txt | 4 + 9 files changed, 407 insertions(+), 19 deletions(-) create mode 100644 tests/conftest.py create mode 100644 tests/invalidTest.txt create mode 100755 tests/test_cdnengine.py create mode 100644 tests/test_detectcdn.py create mode 100644 tests/test_findcdn.py create mode 100644 tests/validTest.txt diff --git a/.isort.cfg b/.isort.cfg index 2369f81..d7aeec1 100644 --- a/.isort.cfg +++ b/.isort.cfg @@ -7,7 +7,7 @@ import_heading_thirdparty=Third-Party Libraries import_heading_firstparty=cisagov Libraries # Should be auto-populated by seed-isort-config hook -known_third_party=dns,docopt,ipwhois,schema,setuptools,tqdm,validators +known_third_party=dns,docopt,ipwhois,pytest,schema,setuptools,tqdm,validators # These must be manually set to correctly separate them from third party libraries known_first_party=findcdn diff --git a/src/findcdn/cdnengine/__init__.py b/src/findcdn/cdnengine/__init__.py index 20bb384..eb69bab 100644 --- a/src/findcdn/cdnengine/__init__.py +++ b/src/findcdn/cdnengine/__init__.py @@ -1,7 +1,7 @@ """cdnengine library.""" # Ignoring mypy because this library does infact exist. from . import detectcdn # type: ignore -from .cdnengine import Chef, DomainPot, run_checks +from .cdnengine import Chef, run_checks # Define public exports -__all__ = ["DomainPot", "Chef", "run_checks", "detectcdn"] +__all__ = ["Chef", "run_checks", "detectcdn"] diff --git a/src/findcdn/cdnengine/cdnengine.py b/src/findcdn/cdnengine/cdnengine.py index ea4cd56..1c62f37 100755 --- a/src/findcdn/cdnengine/cdnengine.py +++ b/src/findcdn/cdnengine/cdnengine.py @@ -21,22 +21,6 @@ from . import detectcdn -class DomainPot: - """DomainPot defines the "pot" which Domain objects are stored.""" - - # pylint: disable=too-few-public-methods - # DomainPot is okay to have only an __init__ as we only need it to have the private var. - - def __init__(self, domains: List[str]): - """Define the pot for the Chef to use.""" - self.domains: List[detectcdn.Domain] = [] - - # Convert to list of type domain - for dom in domains: - dom_in = detectcdn.Domain(dom,) - self.domains.append(dom_in) - - def chef_executor( domain: detectcdn.Domain, timeout: int, diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..39e2be2 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,29 @@ +"""pytest plugin configuration. + +https://docs.pytest.org/en/latest/writing_plugins.html#conftest-py-plugins +""" +# Third-Party Libraries +import pytest + + +def pytest_addoption(parser): + """Add new commandline options to pytest.""" + parser.addoption( + "--runslow", action="store_true", default=False, help="run slow tests" + ) + + +def pytest_configure(config): + """Register new markers.""" + config.addinivalue_line("markers", "slow: mark test as slow") + + +def pytest_collection_modifyitems(config, items): + """Modify collected tests based on custom marks and commandline options.""" + if config.getoption("--runslow"): + # --runslow given in cli: do not skip slow tests + return + skip_slow = pytest.mark.skip(reason="need --runslow option to run") + for item in items: + if "slow" in item.keywords: + item.add_marker(skip_slow) diff --git a/tests/invalidTest.txt b/tests/invalidTest.txt new file mode 100644 index 0000000..6986e0f --- /dev/null +++ b/tests/invalidTest.txt @@ -0,0 +1,17 @@ +notadomain +login.gov +duo.com +www.secureworks.com +status.symantec.com +dnc.org +fitbit.com +zillow.com +ethereum.org +refer.walmart.com +www.netgear.com +view.yahoo.com +today.msnbc.com +www.amazon.co.uk +google.com +facebook.com +asdf.com diff --git a/tests/test_cdnengine.py b/tests/test_cdnengine.py new file mode 100755 index 0000000..3617fd0 --- /dev/null +++ b/tests/test_cdnengine.py @@ -0,0 +1,98 @@ +#!/usr/bin/env pytest -vs +"""Tests for cdnengine.""" + +# Standard Python Libraries + +# cisagov Libraries +import findcdn + +# Test Global Variables +USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/35.0.1916.47 Safari/537.36" +TIMEOUT = 30 +THREADS = 0 # If 0 then cdnengine uses CPU count to set thread count + + +def test_chef_init(): + """Test if Chef can be instantiated correctly.""" + domains = ["asu.edu", "login.gov", "censys.io", "realpython.com"] + chef = findcdn.cdnengine.Chef(domains, THREADS, TIMEOUT, USER_AGENT) + + # Assertions List[detectcdn.Domain] + assert isinstance(chef.pot, list) + + +def test_grab_cdn(): + """Test if Chef can obtain proper CDNs of domains.""" + domains = ["asu.edu", "login.gov", "censys.io", "realpython.com"] + chef = findcdn.cdnengine.Chef(domains, THREADS, TIMEOUT, USER_AGENT) + chef.run_checks() + checked_domains = chef.pot + + # Assertions + assert checked_domains[0].cdns == [ + ".cloudflare.net", + ".cloudflare.com", + ], "Did not detect {} from {}.".format( + [".cloudflare.net", ".cloudflare.com"], checked_domains[0].url + ) + assert checked_domains[1].cdns == [ + ".cloudfront.net", + ], "Did not detect {} from {}.".format([".cloudfront.net"], checked_domains[1].url) + assert checked_domains[2].cdns == [ + ".cloudflare.com" + ], "Did not detect {} from {}.".format([".cloudflare.com"], checked_domains[2].url) + assert checked_domains[3].cdns == [ + ".cloudflare.com" + ], "Did not detect {} from {}.".format([".cloudflare.com"], checked_domains[3].url) + + +def test_has_cdn(): + """Test that of a set of domains with a without a CDN return correctly.""" + domains = ["asu.edu", "censys.io", "bannerhealth.com", "adobe.com"] + chef = findcdn.cdnengine.Chef(domains, THREADS, TIMEOUT, USER_AGENT) + chef.run_checks() + checked_domains = chef.pot + + # Assertions + cdn_present = 0 + for dom in checked_domains: + if len(dom.cdns) > 0: + cdn_present += 1 + + assert cdn_present == 3, "Too many cdn_present domains counted." + assert checked_domains[0].url == "asu.edu" and checked_domains[0].cdns == [ + ".cloudflare.net", + ".cloudflare.com", + ], ("Incorrect CDN detected for %s" % checked_domains[0].url) + assert checked_domains[1].url == "censys.io" and checked_domains[1].cdns == [ + ".cloudflare.com" + ], ("Incorrect CDN detected for %s" % checked_domains[1].url) + + +def test_run_checks(): + """Test the run_checks orchestator works.""" + domains = ["asu.edu", "censys.io", "bannerhealth.com", "adobe.com"] + chef = findcdn.cdnengine.Chef(domains, THREADS, TIMEOUT, USER_AGENT) + chef.run_checks() + + # Assertions + assert len(chef.pot) > 0, "Pot not stored correctly." + + +def test_run_checks_present(): + """Test the return of a list of cdn_present domains.""" + domains = ["asu.edu", "censys.io", "bannerhealth.com", "adobe.com"] + objects, _ = findcdn.cdnengine.run_checks(domains, THREADS, TIMEOUT, USER_AGENT) + cdn_present = {} + for dom in objects: + if len(dom.cdns) > 0: + cdn_present[dom.url] = dom.cdns + expected = { + "asu.edu": [".cloudflare.net", ".cloudflare.com"], + "censys.io": [".cloudflare.com"], + "adobe.com": [".edgekey.net", ".akamaitechnologies.fr"], + } + + # Assertions + assert len(cdn_present) > 0, "Returned cdn_present list is empty." + assert cdn_present == expected, "Returned domains do not match." diff --git a/tests/test_detectcdn.py b/tests/test_detectcdn.py new file mode 100644 index 0000000..ac92576 --- /dev/null +++ b/tests/test_detectcdn.py @@ -0,0 +1,121 @@ +#!/usr/bin/env pytest -vs +"""Tests for detectCDN.""" + +# Third-Party Libraries +import dns.resolver + +# cisagov Libraries +from findcdn.cdnengine.detectcdn import ( + Domain, + all_checks, + cname, + https_lookup, + ip_resolve, + whois, +) + +# Globals +TIMEOUT = 10 +USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/35.0.1916.47 Safari/537.36" + + +def test_ip(): + """Test the IP resolving feature.""" + dns.resolver.default_resolver = dns.resolver.Resolver() + dns.resolver.default_resolver.nameservers = ["1.1.1.1", "8.8.8.8"] + dom_in = Domain("dns.google.com") + ip_resolve(dom_in) + + assert "8.8.8.8" in dom_in.ip_list, "the ip for dns.google.com should be 8.8.8.8" + + +def test_broken_ip(): + """Test a non-working domain IP resolving feature.""" + dns.resolver.default_resolver = dns.resolver.Resolver() + dns.resolver.default_resolver.nameservers = ["1.1.1.1", "8.8.8.8"] + dom_in = Domain("notarealdomain.fakedomaindne.com") + return_code = ip_resolve(dom_in) + assert return_code != 0, "This fake site should return a non 0 code." + + +def test_cname(): + """Test the CNAME resolving feature.""" + dom_in = Domain("www.asu.edu") + cname(dom_in, timeout=TIMEOUT) + + assert ( + "www.asu.edu.cdn.cloudflare.net." in dom_in.cnames + ), "www.asu.edu should have www.asu.edu.cdn.cloudflare.net. as a cname" + + +def test_broken_cname(): + """Test a non-working domain CNAME resolving feature.""" + dom_in = Domain("notarealdomain.fakedomaindne.com",) + return_code = cname(dom_in, timeout=TIMEOUT) + assert return_code != 0, "This fake site should return a non 0 code." + + +def test_https_lookup(): + """Test the header resolving feature.""" + dom_in = Domain("google.com",) + https_lookup(dom_in, timeout=TIMEOUT, agent=USER_AGENT) + + assert "gws" in dom_in.headers, "google.com should have gws as a header" + + +def test_broken_https_lookup(): + """Test a non-working domain header resolving feature.""" + dom_in = Domain("notarealdomain.fakedomaindne.com",) + https_lookup(dom_in, timeout=TIMEOUT, agent=USER_AGENT) + assert len(dom_in.headers) <= 0, "There should be no response." + + +def test_whois(): + """Test the whois resolving feature.""" + dom_in = Domain("google.com") + ip_resolve(dom_in) + whois(dom_in) + + assert ( + "GOOGLE" in dom_in.whois_data + ), "google.com should return GOOGLE in the whois_data" + + +def test_broken_whois(): + """Test a non-working domain whois resolving feature.""" + dom_in = Domain("notarealdomain.fakedomaindne.com",) + ip_resolve(dom_in) + return_code = whois(dom_in) + assert return_code != 0, "This fake site should return a non 0 code." + + +def test_all_checks(): + """Run all checks.""" + dom_in = Domain("login.gov") + all_checks(dom_in, timeout=TIMEOUT, agent=USER_AGENT) + + assert ( + ".cloudfront.net" in dom_in.cdns + ), "the ip for dns.google.com should be 8.8.8.8" + + +def test_all_checks_by_name(): + """Run all checks and get CDN name.""" + dom_in = Domain("login.gov") + all_checks(dom_in, timeout=TIMEOUT, agent=USER_AGENT) + + assert ( + ".cloudfront.net" in dom_in.cdns + ), "the ip for dns.google.com should be 8.8.8.8" + + +def test_all_checks_bad(): + """Test fake domain and ensure it dosen't break anything.""" + dns.resolver.default_resolver = dns.resolver.Resolver() + dns.resolver.default_resolver.nameservers = ["1.1.1.1", "8.8.8.8"] + dom = Domain("super.definitelynot.notarealdomain.fakedomaindne.com",) + print(dom.url, dom.cdns, dom.cnames, dom.headers, dom.whois_data, dom.ip_list) + return_code = all_checks(dom, timeout=TIMEOUT, agent=USER_AGENT) + print(return_code) + print(dom.url, dom.cdns, dom.cnames, dom.headers, dom.whois_data, dom.ip_list) + assert return_code != 0, "This fake site should return a non 0 code." diff --git a/tests/test_findcdn.py b/tests/test_findcdn.py new file mode 100644 index 0000000..4f7a7f5 --- /dev/null +++ b/tests/test_findcdn.py @@ -0,0 +1,135 @@ +#!/usr/bin/env pytest -vs +"""Tests for findcdn.""" + +# Standard Python Libraries +import os +import sys +from unittest.mock import patch + +# Third-Party Libraries +import pytest + +# cisagov Libraries +import findcdn + +# define sources of version strings +RELEASE_TAG = os.getenv("RELEASE_TAG") +PROJECT_VERSION = findcdn.__version__ + + +def test_stdout_version(capsys): + """Verify that version string sent to stdout agrees with the module version.""" + with pytest.raises(SystemExit): + with patch.object(sys, "argv", ["bogus", "--version"]): + findcdn.findcdn.interactive() + captured = capsys.readouterr() + assert ( + captured.out == f"{PROJECT_VERSION}\n" + ), "standard output by '--version' should agree with module.__version__" + + +@pytest.mark.skipif( + RELEASE_TAG in [None, ""], reason="this is not a release (RELEASE_TAG not set)" +) +def test_release_version(): + """Verify that release tag version agrees with the module version.""" + assert ( + RELEASE_TAG == f"v{PROJECT_VERSION}" + ), "RELEASE_TAG does not match the project version" + + +def test_list_working(): + """Working domain list to test with.""" + with patch.object( + sys, "argv", ["bogus", "list", "google.com", "facebook.com", "login.gov"] + ): + return_code = findcdn.findcdn.interactive() + assert return_code == 0, "interactive() should return successfully" + + +def test_list_working_double(capsys): + """Working domain list to test -d with.""" + with patch.object( + sys, + "argv", + ["bogus", "list", "google.com", "facebook.com", "login.gov", "-v", "-d"], + ): + return_code = findcdn.findcdn.interactive() + captured = capsys.readouterr() + assert return_code == 0, "interactive() should return successfully" + assert "6 jobs completed" in captured.out + + +def test_list_working_verbose(capsys): + """Working domain list to test -v with.""" + with patch.object( + sys, + "argv", + [ + "bogus", + "list", + "google.com", + "facebook.com", + "superfake.thisisnotarealwebsiteforsure.com", + "-v", + ], + ): + return_code = findcdn.findcdn.interactive() + captured = capsys.readouterr() + assert return_code == 0, "interactive() should return successfully" + assert "3 Domains Validated" in captured.out + + +def test_list_working_tcount(capsys): + """Working domain list to test -t with.""" + with patch.object( + sys, + "argv", + ["bogus", "list", "google.com", "facebook.com", "login.gov", "-t", "3", "-v"], + ): + return_code = findcdn.findcdn.interactive() + captured = capsys.readouterr() + assert return_code == 0, "interactive() should return successfully" + assert "Using 3 threads" in captured.out + + +def test_list_broken(): + """Broken domain list to test with.""" + with patch.object( + sys, + "argv", + ["bogus", "list", "google.com/searchtest", "facebook.com", "login.gov"], + ): + return_code = findcdn.findcdn.interactive() + assert return_code == 3, "interactive() should return failure" + + +def test_file_working(): + """Working domain file to test with.""" + with patch.object(sys, "argv", ["./findcdn", "file", "tests/validTest.txt"]): + return_code = findcdn.findcdn.interactive() + assert return_code == 0, "interactive() should return successfully" + + +def test_file_broken(): + """Broken domain file to test with.""" + with patch.object(sys, "argv", ["bogus", "file", "tests/invalidTest.txt"]): + return_code = findcdn.findcdn.interactive() + assert return_code != 0, "interactive() should return failure" + + +def test_file_dne(): + """Working domain list to test with.""" + with patch.object(sys, "argv", ["./findcdn", "file", "nosuchfile.txt"]): + return_code = findcdn.findcdn.interactive() + assert return_code != 0, "interactive() should return successfully" + + +def test_file_write(tmpdir): + """Test writing to a file.""" + file = tmpdir.join("outputtest.txt") + with patch.object( + sys, "argv", ["./findcdn", "list", "google.com", "-o", str(file)] + ): + findcdn.findcdn.interactive() + assert "google.com" in file.read() diff --git a/tests/validTest.txt b/tests/validTest.txt new file mode 100644 index 0000000..7288725 --- /dev/null +++ b/tests/validTest.txt @@ -0,0 +1,4 @@ +login.gov +duo.com +facebook.com +asdf.com From a11c66708586ca15d5b6ee8b7dfcd422cce77459 Mon Sep 17 00:00:00 2001 From: Pascal-0x90 Date: Mon, 10 Aug 2020 10:33:52 -0700 Subject: [PATCH 5/8] Fixed commend and removed lambda for pull #16. Signed-off-by: Pascal-0x90 --- src/findcdn/cdnengine/cdnengine.py | 3 ++- src/findcdn/findcdn.py | 4 +--- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/src/findcdn/cdnengine/cdnengine.py b/src/findcdn/cdnengine/cdnengine.py index 1c62f37..3faa254 100755 --- a/src/findcdn/cdnengine/cdnengine.py +++ b/src/findcdn/cdnengine/cdnengine.py @@ -140,10 +140,11 @@ def grab_cdn( # Update status bar if allowed if self.pbar: - # We type ignore these as its "illegal" to access private attributes of an object but we need to (thanks python) + # Store values in format strings to then set in progress bar pending = f"Pending: {job_count - completed} jobs" threads = f"Threads: {self.threads}" pbar.set_description(f"[{pending}]==[{threads}]") + # If we have a progress bar, increment it if self.pbar is not None: pbar.update(1) completed += 1 diff --git a/src/findcdn/findcdn.py b/src/findcdn/findcdn.py index df50205..b45ee5a 100755 --- a/src/findcdn/findcdn.py +++ b/src/findcdn/findcdn.py @@ -167,9 +167,7 @@ def interactive() -> int: None, And( str, - lambda filename: os.path.isfile( # pylint: disable=unnecessary-lambda - filename - ), + os.path.isfile(args[""]), error='Input file "' + str(args[""]) + '" does not exist!', ), ), From 38e77d52235c3a7f8bb942456bee1f3772b4a8ec Mon Sep 17 00:00:00 2001 From: Pascal-0x90 Date: Mon, 10 Aug 2020 10:56:36 -0700 Subject: [PATCH 6/8] Fixed test issue Signed-off-by: Pascal-0x90 --- src/findcdn/findcdn.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/findcdn/findcdn.py b/src/findcdn/findcdn.py index b45ee5a..7901f79 100755 --- a/src/findcdn/findcdn.py +++ b/src/findcdn/findcdn.py @@ -167,7 +167,7 @@ def interactive() -> int: None, And( str, - os.path.isfile(args[""]), + lambda filename: os.path.isfile(str(filename)), error='Input file "' + str(args[""]) + '" does not exist!', ), ), From c4daf236a74dcdd6e0f5b5e0110fc07dabaa8a6e Mon Sep 17 00:00:00 2001 From: Nathan <35417308+Pascal-0x90@users.noreply.github.com> Date: Wed, 12 Aug 2020 07:56:04 -0700 Subject: [PATCH 7/8] Update cdn_check.py Removed comment about timeout and explained use of \#nosec --- src/findcdn/cdnengine/detectcdn/cdn_check.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/findcdn/cdnengine/detectcdn/cdn_check.py b/src/findcdn/cdnengine/detectcdn/cdn_check.py index e413749..86e3ef9 100755 --- a/src/findcdn/cdnengine/detectcdn/cdn_check.py +++ b/src/findcdn/cdnengine/detectcdn/cdn_check.py @@ -135,7 +135,8 @@ def https_lookup(dom: Domain, timeout: int, agent: str) -> int: req = request.Request( protocol + dom.url, data=None, headers={"User-Agent": agent}, ) - # Making the timeout 50 as to not hang thread. + # The use of urlopen is safe in this context as we do not let the user + # have choice over the use of what protocol urlopen() uses. response = request.urlopen(req, timeout=timeout) # nosec except ( HTTPError, From 968bbb69bf91753edbbd109531d2b1c1c72f87de Mon Sep 17 00:00:00 2001 From: Pascal-0x90 Date: Wed, 12 Aug 2020 08:00:18 -0700 Subject: [PATCH 8/8] Fixing github not fixing my spacing. Signed-off-by: Pascal-0x90 --- src/findcdn/cdnengine/detectcdn/cdn_check.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/findcdn/cdnengine/detectcdn/cdn_check.py b/src/findcdn/cdnengine/detectcdn/cdn_check.py index 86e3ef9..1500027 100755 --- a/src/findcdn/cdnengine/detectcdn/cdn_check.py +++ b/src/findcdn/cdnengine/detectcdn/cdn_check.py @@ -136,7 +136,7 @@ def https_lookup(dom: Domain, timeout: int, agent: str) -> int: protocol + dom.url, data=None, headers={"User-Agent": agent}, ) # The use of urlopen is safe in this context as we do not let the user - # have choice over the use of what protocol urlopen() uses. + # have choice over the use of what protocol urlopen() uses. response = request.urlopen(req, timeout=timeout) # nosec except ( HTTPError,