IDEV-1995: Add domainrdap cli

Author: jbabac

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).",
 }