IDEV-1995: implement parsed domain rdap feed

domaintools/api.py

@@ -1085,3 +1085,18 @@ def nad(self, **kwargs):
             response_path=(),
             **kwargs,
         )
+
+    def domainrdap(self, **kwargs):
+        """Returns changes to global domain registration information, populated by the Registration Data Access Protocol (RDAP)"""
+        sessionID = kwargs.get("sessionID")
+        after = kwargs.get("after")
+        before = kwargs.get("before")
+        if not (sessionID or after or before):
+            raise ValueError("sessionID or after or before must be defined")
+
+        return self._results(
+            "domain-registration-data-access-protocol-feed-(api)",
+            "v1/feed/domainrdap/",
+            response_path=(),
+            **kwargs,
+        )

domaintools/base_results.py

@@ -138,7 +138,7 @@ def data(self):
             return self._data
 
     def check_limit_exceeded(self):
-        if self.kwargs.get("format", "json") == "json":
+        if self.kwargs.get("format", "json") == "json" and self.product not in get_feeds_products_list():
             if "response" in self._data and "limit_exceeded" in self._data["response"] and self._data["response"]["limit_exceeded"] is True:
                 return True, self._data["response"]["message"]
         # TODO: handle html, xml response errors better.

domaintools/cli/api.py

@@ -4,6 +4,7 @@
 import os
 import _io
 
+from datetime import datetime
 from typing import Optional, Dict, Tuple
 from rich.progress import Progress, SpinnerColumn, TextColumn
 
@@ -28,11 +29,21 @@ def print_api_version(value: bool):
     def validate_format_input(value: str):
         VALID_FORMATS = ("list", "json", "xml", "html")
         if value not in VALID_FORMATS:
-            raise typer.BadParameter(
-                f"{value} is not in available formats: {VALID_FORMATS}"
-            )
+            raise typer.BadParameter(f"{value} is not in available formats: {VALID_FORMATS}")
         return value
 
+    @staticmethod
+    def validate_after_or_before_input(value: str):
+        if value is None or value.replace("-", "").isdigit():
+            return value
+
+        # Checks if value is a valid ISO 8601 datetime string in UTC form
+        try:
+            datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ")
+            return value
+        except:
+            raise typer.BadParameter(f"{value} is neither an integer or a valid ISO 8601 datetime string in UTC form")
+
     @staticmethod
     def validate_source_file_extension(value: str):
         """Validates source file extension.
@@ -51,9 +62,7 @@ def validate_source_file_extension(value: str):
         ext = get_file_extension(value)
 
         if ext.lower() not in VALID_EXTENSIONS:
-            raise typer.BadParameter(
-                f"{value} is not in valid extensions. Valid file extensions: {VALID_EXTENSIONS}"
-            )
+            raise typer.BadParameter(f"{value} is not in valid extensions. Valid file extensions: {VALID_EXTENSIONS}")
 
         return value
 
@@ -85,11 +94,7 @@ def args_to_dict(*args) -> Dict:
     def _get_formatted_output(cls, cmd_name: str, response, out_format: str = "json"):
         if cmd_name in ("available_api_calls",):
             return "\n".join(response)
-        return str(
-            getattr(response, out_format)
-            if out_format != "list"
-            else response.as_list()
-        )
+        return str(getattr(response, out_format) if out_format != "list" else response.as_list())
 
     @classmethod
     def _get_credentials(cls, params: Optional[Dict] = {}) -> Tuple[str]:
@@ -106,9 +111,7 @@ def _get_credentials(cls, params: Optional[Dict] = {}) -> Tuple[str]:
                 with open(creds_file, "r") as cf:
                     user, key = cf.readline().strip(), cf.readline().strip()
             except FileNotFoundError as e:
-                raise typer.BadParameter(
-                    f"{str(e)}. Please create one first and try again."
-                )
+                raise typer.BadParameter(f"{str(e)}. Please create one first and try again.")
 
         return user, key
 
@@ -198,9 +201,7 @@ def run(cls, name: str, params: Optional[Dict] = {}, **kwargs):
                     total=None,
                 )
 
-                output = cls._get_formatted_output(
-                    cmd_name=name, response=response, out_format=response_format
-                )
+                output = cls._get_formatted_output(cmd_name=name, response=response, out_format=response_format)
 
                 if isinstance(out_file, _io.TextIOWrapper):
                     # use rich `print` command to prettify the ouput in sys.stdout
@@ -215,10 +216,7 @@ def run(cls, name: str, params: Optional[Dict] = {}, **kwargs):
                 _reason = getattr(e, "reason", {})
                 # check data type first as some of the reasons is just plain text
                 if isinstance(_reason, dict):
-                    _reason = (
-                        _reason.get("error", {}).get("message")
-                        or "Unknown Error occured."
-                    )
+                    _reason = _reason.get("error", {}).get("message") or "Unknown Error occured."
 
                 reason = typer.style(_reason, bg=typer.colors.RED)
 

domaintools/cli/commands/feeds.py

@@ -35,9 +35,7 @@ def feeds_nad(
         help="Output format in {'list', 'json', 'xml', 'html'}",
         callback=DTCLICommand.validate_format_input,
     ),
-    out_file: typer.FileTextWrite = typer.Option(
-        sys.stdout, "-o", "--out-file", help="Output file (defaults to stdout)"
-    ),
+    out_file: typer.FileTextWrite = typer.Option(sys.stdout, "-o", "--out-file", help="Output file (defaults to stdout)"),
     no_verify_ssl: bool = typer.Option(
         False,
         "--no-verify-ssl",
@@ -100,9 +98,7 @@ def feeds_nod(
         help="Output format in {'list', 'json', 'xml', 'html'}",
         callback=DTCLICommand.validate_format_input,
     ),
-    out_file: typer.FileTextWrite = typer.Option(
-        sys.stdout, "-o", "--out-file", help="Output file (defaults to stdout)"
-    ),
+    out_file: typer.FileTextWrite = typer.Option(sys.stdout, "-o", "--out-file", help="Output file (defaults to stdout)"),
     no_verify_ssl: bool = typer.Option(
         False,
         "--no-verify-ssl",
@@ -136,3 +132,59 @@ def feeds_nod(
     ),
 ):
     DTCLICommand.run(name=c.FEEDS_NOD, params=ctx.params)
+
+
+@dt_cli.command(
+    name=c.FEEDS_DOMAINRDAP,
+    help=get_cli_helptext_by_name(command_name=c.FEEDS_DOMAINRDAP),
+)
+def feeds_domainrdap(
+    ctx: typer.Context,
+    user: str = typer.Option(None, "-u", "--user", help="Domaintools API Username."),
+    key: str = typer.Option(None, "-k", "--key", help="DomainTools API key"),
+    creds_file: str = typer.Option(
+        "~/.dtapi",
+        "-c",
+        "--credfile",
+        help="Optional file with API username and API key, one per line.",
+    ),
+    no_verify_ssl: bool = typer.Option(
+        False,
+        "--no-verify-ssl",
+        help="Skip verification of SSL certificate when making HTTPs API calls",
+    ),
+    no_sign_api_key: bool = typer.Option(
+        False,
+        "--no-sign-api-key",
+        help="Skip signing of api key",
+    ),
+    sessionID: str = typer.Option(
+        None,
+        "--session-id",
+        help="Unique identifier for the session",
+    ),
+    after: str = typer.Option(
+        None,
+        "--after",
+        help="Start of the time window, relative to the current time in seconds, for which data will be provided",
+        callback=DTCLICommand.validate_after_or_before_input,
+    ),
+    before: str = typer.Option(
+        None,
+        "--before",
+        help="The end of the query window in seconds, relative to the current time, inclusive",
+        callback=DTCLICommand.validate_after_or_before_input,
+    ),
+    domain: str = typer.Option(
+        None,
+        "-d",
+        "--domain",
+        help="A string value used to filter feed results",
+    ),
+    top: str = typer.Option(
+        None,
+        "--top",
+        help="Number of results to return in the response payload",
+    ),
+):
+    DTCLICommand.run(name=c.FEEDS_DOMAINRDAP, params=ctx.params)

domaintools/cli/constants.py

@@ -46,3 +46,4 @@
 # feeds
 FEEDS_NAD = "nad"
 FEEDS_NOD = "nod"
+FEEDS_DOMAINRDAP = "domainrdap"

domaintools/cli/utils.py

@@ -84,6 +84,7 @@ def _phisheye_termlist():
     c.IRIS_DETECT_IGNORED_DOMAINS: "Returns back a list of ignored domains in Iris Detect based on the provided filters.",
     c.FEEDS_NAD: "Returns back newly active domains feed.",
     c.FEEDS_NOD: "Returns back newly observed domains feed.",
+    c.FEEDS_DOMAINRDAP: "Returns changes to global domain registration information, populated by the Registration Data Access Protocol (RDAP).",
 }
 
 

domaintools/utils.py

@@ -19,9 +19,7 @@ def get_domain_age(create_date):
         try:
             create_date = datetime.strptime(create_date, "%Y%m%d")
         except ValueError:
-            raise ValueError(
-                "Invalid date format. Supported formats are %Y-%m-%d and %Y%m%d."
-            )
+            raise ValueError("Invalid date format. Supported formats are %Y-%m-%d and %Y%m%d.")
 
     time_diff = datetime.now() - create_date
 
@@ -110,11 +108,7 @@ def prune_data(data_obj):
             prune_data(item)
             if not isinstance(item, int) and not item:
                 items_to_prune.append(index)
-        data_obj[:] = [
-            item
-            for index, item in enumerate(data_obj)
-            if index not in items_to_prune and len(item)
-        ]
+        data_obj[:] = [item for index, item in enumerate(data_obj) if index not in items_to_prune and len(item)]
 
 
 def find_emails(data_str):
@@ -151,9 +145,7 @@ def get_pivots(data_obj, name, return_data=None, count=0, pivot_threshold=500):
         for k, v in data_obj.items():
             if isinstance(data_obj[k], (dict, list)):
                 name = "{}_{}".format(name, k)
-                temp_data = get_pivots(
-                    data_obj[k], name, return_data, count, pivot_threshold
-                )
+                temp_data = get_pivots(data_obj[k], name, return_data, count, pivot_threshold)
                 if temp_data:
                     return_data.append([name[1:].upper().replace("_", " "), temp_data])
             name = temp_name
@@ -175,14 +167,13 @@ def get_pivots(data_obj, name, return_data=None, count=0, pivot_threshold=500):
         return return_data
 
 
-def convert_str_to_dateobj(
-    string_date: str, date_format: Optional[str] = "%Y-%m-%d"
-) -> datetime:
+def convert_str_to_dateobj(string_date: str, date_format: Optional[str] = "%Y-%m-%d") -> datetime:
     return datetime.strptime(string_date, date_format)
 
 
 def get_feeds_products_list():
     return [
         "newly-active-domains-feed-(api)",
         "newly-observed-domains-feed-(api)",
+        "domain-registration-data-access-protocol-feed-(api)",
     ]