IDEV-1996: Implement domaindiscovery cli.

Author: jbabac

domaintools/cli/api.py

@@ -8,6 +8,7 @@
 from typing import Optional, Dict, Tuple
 from rich.progress import Progress, SpinnerColumn, TextColumn
 
+from domaintools.constants import Endpoint, OutputFormat
 from domaintools.api import API
 from domaintools.exceptions import ServiceException
 from domaintools.cli.utils import get_file_extension
@@ -32,6 +33,20 @@ def validate_format_input(value: str):
             raise typer.BadParameter(f"{value} is not in available formats: {VALID_FORMATS}")
         return value
 
+    @staticmethod
+    def validate_feeds_format_input(value: str):
+        VALID_FEEDS_FORMATS = ("jsonl", "csv")
+        if value not in VALID_FEEDS_FORMATS:
+            raise typer.BadParameter(f"{value} is not in available formats: {VALID_FEEDS_FORMATS}")
+        return value
+
+    @staticmethod
+    def validate_endpoint_input(value: str):
+        VALID_ENDPOINTS = (Endpoint.FEED.value, Endpoint.DOWNLOAD.value)
+        if value not in VALID_ENDPOINTS:
+            raise typer.BadParameter(f"{value} is not in available endpoints: {VALID_ENDPOINTS}")
+        return value
+
     @staticmethod
     def validate_after_or_before_input(value: str):
         if value is None or value.replace("-", "").isdigit():
@@ -152,7 +167,13 @@ def run(cls, name: str, params: Optional[Dict] = {}, **kwargs):
         """
         try:
             rate_limit = params.pop("rate_limit", False)
-            response_format = params.pop("format", "json")
+            response_format = (
+                params.pop("format", "json")
+                if params.get("format", None)
+                else params.get(
+                    "output_format", OutputFormat.JSONL.value
+                )  # Using output_format for RTUF endpoints to separate from other endpoints. This will be needed further along the process
+            )
             out_file = params.pop("out_file", sys.stdout)
             verify_ssl = params.pop("no_verify_ssl", False)
             always_sign_api_key = params.pop("no_sign_api_key", False)

domaintools/cli/commands/feeds.py

@@ -6,6 +6,7 @@
 from domaintools.cli.api import DTCLICommand
 from domaintools.cli.utils import get_cli_helptext_by_name
 from domaintools.cli import constants as c
+from domaintools.constants import Endpoint
 
 
 @dt_cli.command(
@@ -158,6 +159,13 @@ def feeds_domainrdap(
         "--no-sign-api-key",
         help="Skip signing of api key",
     ),
+    endpoint: str = typer.Option(
+        Endpoint.FEED.value,
+        "-e",
+        "--endpoint",
+        help=f"Valid endpoints: [{Endpoint.FEED.value}, {Endpoint.DOWNLOAD.value}]",
+        callback=DTCLICommand.validate_endpoint_input,
+    ),
     sessionID: str = typer.Option(
         None,
         "--session-id",
@@ -188,3 +196,78 @@ def feeds_domainrdap(
     ),
 ):
     DTCLICommand.run(name=c.FEEDS_DOMAINRDAP, params=ctx.params)
+
+
+@dt_cli.command(
+    name=c.FEEDS_DOMAINDISCOVERY,
+    help=get_cli_helptext_by_name(command_name=c.FEEDS_DOMAINDISCOVERY),
+)
+def feeds_domaindiscovery(
+    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",
+    ),
+    output_format: str = typer.Option(
+        "jsonl",
+        "-f",
+        "--format",
+        help="Output format in {'jsonl'}",
+        callback=DTCLICommand.validate_feeds_format_input,
+    ),
+    endpoint: str = typer.Option(
+        Endpoint.FEED.value,
+        "-e",
+        "--endpoint",
+        help=f"Valid endpoints: [{Endpoint.FEED.value}, {Endpoint.DOWNLOAD.value}]",
+        callback=DTCLICommand.validate_endpoint_input,
+    ),
+    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",
+    ),
+    headers: bool = typer.Option(
+        False,
+        "--headers",
+        help="Adds a header to the first line of response when text/csv is set in header parameters",
+    ),
+    top: str = typer.Option(
+        None,
+        "--top",
+        help="Number of results to return in the response payload. This is ignored in download endpoint",
+    ),
+):
+    DTCLICommand.run(name=c.FEEDS_DOMAINDISCOVERY, params=ctx.params)

domaintools/cli/constants.py

@@ -47,3 +47,4 @@
 FEEDS_NAD = "nad"
 FEEDS_NOD = "nod"
 FEEDS_DOMAINRDAP = "domainrdap"
+FEEDS_DOMAINDISCOVERY = "domaindiscovery"