feat(cli): inventory search and saved queries from cli

censys/asm/api.py

@@ -110,3 +110,11 @@ def _get_logbook_page(
             args = {"cursor": res["nextCursor"]}
 
             yield from res["events"]
+
+    def get_workspace_id(self) -> str:
+        """Get the workspace ID.
+
+        Returns:
+            str: The workspace ID.
+        """
+        return self._get("/integrations/v1/account")["workspaceId"]

censys/asm/inventory.py

@@ -11,7 +11,7 @@ class InventorySearch(CensysAsmAPI):
 
     def search(
         self,
-        workspaces: List[str],
+        workspaces: Optional[List[str]] = None,
         query: Optional[str] = None,
         page_size: Optional[int] = None,
         cursor: Optional[str] = None,
@@ -21,7 +21,7 @@ def search(
         """Search inventory data.
 
         Args:
-            workspaces (List[str]): List of workspace IDs to search.
+            workspaces (List[str], optional): List of workspace IDs to search. Deprecated. The workspace associated with `CENSYS-API-KEY` will be used automatically.
             query (str, optional): Query string.
             page_size (int, optional): Number of results to return. Defaults to 50.
             cursor (str, optional): Cursor to start search from.
@@ -31,17 +31,29 @@ def search(
         Returns:
             dict: Inventory search results.
         """
+        if workspaces is not None:
+            print(
+                "The field 'workspaces' is being deprecated. The workspace associated with `CENSYS-API-KEY` will be used automatically."
+            )
         if page_size is None:
             page_size = 50
+        w: List[str] = []
+        w.append(self.get_workspace_id())
+
         args = {
-            "workspaces": workspaces,
-            "query": query,
+            "workspaces": w,
             "pageSize": page_size,
-            "cursor": cursor,
-            "sort": sort,
-            "fields": fields,
         }
 
+        if query:
+            args["query"] = query
+        if cursor:
+            args["cursor"] = cursor
+        if sort:
+            args["sort"] = sort
+        if fields:
+            args["fields"] = fields
+
         return self._get(self.base_path, args=args)
 
     def aggregate(

censys/cli/commands/asm.py

@@ -11,6 +11,7 @@
 from rich.progress import Progress, TaskID
 from rich.prompt import Confirm, Prompt
 
+from censys.asm.inventory import InventorySearch
 from censys.asm.saved_queries import SavedQueries
 from censys.asm.seeds import SEED_TYPES, Seeds
 from censys.cli.utils import console
@@ -530,6 +531,74 @@ def cli_delete_saved_query_by_id(args: argparse.Namespace):
         sys.exit(1)
 
 
+def cli_execute_saved_query_by_name(args: argparse.Namespace):
+    """Execute saved query by name subcommand.
+
+    Args:
+        args (Namespace): Argparse Namespace.
+    """
+    s = InventorySearch(args.api_key)
+    q = SavedQueries(args.api_key)
+    # get query from name
+    queries = q.get_saved_queries(args.query_name, 1, 1)
+    if not queries["results"]:
+        console.print("No saved query found with that name.")
+        sys.exit(1)
+    query = queries["results"][0]["query"]
+
+    try:
+        res = s.search(None, query, args.page_size, None, args.sort, args.fields)
+        console.print_json(json.dumps(res))
+    except (KeyError, CensysAsmException):
+        console.print("Failed to execute saved query.")
+        sys.exit(1)
+
+
+def cli_execute_saved_query_by_id(args: argparse.Namespace):
+    """Execute saved query by id subcommand.
+
+    Args:
+        args (Namespace): Argparse Namespace.
+    """
+    s = InventorySearch(args.api_key)
+    q = SavedQueries(args.api_key)
+    try:
+        query_json = q.get_saved_query_by_id(args.query_id)
+        query = query_json["result"]["query"]
+    except (KeyError, CensysAsmException):
+        console.print("No saved query found with that ID.")
+        sys.exit(1)
+    try:
+        res = s.search(None, query, args.page_size, None, args.sort, args.fields)
+        console.print_json(json.dumps(res))
+    except (KeyError, CensysAsmException):
+        console.print("Failed to execute saved query.")
+        sys.exit(1)
+
+
+def cli_search(args: argparse.Namespace):
+    """Inventory search subcommand.
+
+    Args:
+        args (Namespace): Argparse Namespace.
+    """
+    s = InventorySearch(args.api_key)
+
+    try:
+        res = s.search(
+            args.workspaces,
+            args.query,
+            args.page_size,
+            args.cursor,
+            args.sort,
+            args.fields,
+        )
+        console.print_json(json.dumps(res))
+    except (KeyError, CensysAsmException):
+        console.print("Failed to execute query.")
+        sys.exit(1)
+
+
 def include(parent_parser: argparse._SubParsersAction, parents: dict):
     """Include this subcommand into the parent parser.
 
@@ -775,3 +844,116 @@ def add_verbose(parser):
     )
     add_verbose(delete_saved_query_by_id_parser)
     delete_saved_query_by_id_parser.set_defaults(func=cli_delete_saved_query_by_id)
+
+    execute_saved_query_by_name_parser = asm_subparser.add_parser(
+        "execute-saved-query-by-name",
+        description="Execute a saved query by name in inventory search",
+        help="execute saved query by name",
+        parents=[parents["asm_auth"]],
+    )
+    execute_saved_query_by_name_parser.add_argument(
+        "--query-name",
+        help="Query name",
+        type=str,
+        required=True,
+    )
+    execute_saved_query_by_name_parser.add_argument(
+        "--page-size",
+        help="Number of results to return. Defaults to 50.",
+        type=int,
+        default=50,
+    )
+    execute_saved_query_by_name_parser.add_argument(
+        "--sort",
+        help="Sort order for results",
+        type=List[str],
+        default=[],
+    )
+    execute_saved_query_by_name_parser.add_argument(
+        "--fields",
+        help="Fields to include in results",
+        type=List[str],
+        default=[],
+    )
+    add_verbose(execute_saved_query_by_name_parser)
+    execute_saved_query_by_name_parser.set_defaults(
+        func=cli_execute_saved_query_by_name
+    )
+
+    execute_saved_query_by_id_parser = asm_subparser.add_parser(
+        "execute-saved-query-by-id",
+        description="Execute a saved query by id in inventory search",
+        help="execute saved query by id",
+        parents=[parents["asm_auth"]],
+    )
+    execute_saved_query_by_id_parser.add_argument(
+        "--query-id",
+        help="Query ID",
+        type=str,
+        required=True,
+    )
+    execute_saved_query_by_id_parser.add_argument(
+        "--page-size",
+        help="Number of results to return. Defaults to 50.",
+        type=int,
+        default=50,
+    )
+    execute_saved_query_by_id_parser.add_argument(
+        "--sort",
+        help="Sort order for results",
+        type=List[str],
+        default=[],
+    )
+    execute_saved_query_by_id_parser.add_argument(
+        "--fields",
+        help="Fields to include in results",
+        type=List[str],
+        default=[],
+    )
+    add_verbose(execute_saved_query_by_id_parser)
+    execute_saved_query_by_id_parser.set_defaults(func=cli_execute_saved_query_by_id)
+
+    search_parser = asm_subparser.add_parser(
+        "search",
+        description="Execute a query in inventory search",
+        help="execute query in inventory search",
+        parents=[parents["asm_auth"]],
+    )
+    search_parser.add_argument(
+        "--query",
+        help="Query string",
+        type=str,
+        required=True,
+    )
+    search_parser.add_argument(
+        "--page-size",
+        help="Number of results to return. Defaults to 50.",
+        type=int,
+        default=50,
+    )
+    search_parser.add_argument(
+        "--cursor",
+        help="Cursor to use for pagination",
+        type=str,
+        default="",
+    )
+    search_parser.add_argument(
+        "--sort",
+        help="Sort order for results",
+        type=List[str],
+        default=[],
+    )
+    search_parser.add_argument(
+        "--fields",
+        help="Fields to include in results",
+        type=List[str],
+        default=[],
+    )
+    search_parser.add_argument(
+        "--workspaces",
+        help="Workspace IDs to search. Deprecated. The workspace associated with `CENSY-API-KEY` will be used automatically.",
+        type=str,
+        required=False,
+    )
+    add_verbose(search_parser)
+    search_parser.set_defaults(func=cli_search)

tests/asm/test_api.py

@@ -108,3 +108,17 @@ def keyword_callback(request):
         res = list(self.api._get_page(f"/{keyword}"))
         # Assertions
         assert res == page_json[keyword] + second_page
+
+    def test_get_workspace_id(self):
+        self.responses.add(
+            responses.GET,
+            f"{self.base_url}/integrations/v1/account",
+            status=200,
+            json={"workspaceId": "test-workspace-id"},
+        )
+
+        # Actual call
+        res = self.api.get_workspace_id()
+
+        # Assertions
+        assert res == "test-workspace-id"

tests/asm/test_inventory.py

@@ -2,7 +2,7 @@
 from parameterized import parameterized
 
 from ..utils import CensysTestCase
-from .utils import BASE_URL
+from .utils import BASE_URL, WORKSPACE_ID
 from censys.asm.inventory import InventorySearch
 
 INVENTORY_BASE_PATH = f"{BASE_URL}/inventory/v1"
@@ -47,25 +47,27 @@ def setUp(self):
         [
             (
                 {
-                    "workspaces": ["1", "2"],
                     "query": "test",
                     "page_size": 50,
                     "cursor": "test",
                     "sort": ["test"],
                     "fields": ["test"],
                 },
-                "?workspaces=1&workspaces=2&query=test&pageSize=50&cursor=test&sort=test&fields=test",
+                "?workspaces=test-workspace-id&query=test&pageSize=50&cursor=test&sort=test&fields=test",
             ),
             (
                 {
                     "workspaces": ["1", "2"],
                     "query": "test",
                 },
-                "?workspaces=1&workspaces=2&query=test&pageSize=50",
+                "?workspaces=test-workspace-id&query=test&pageSize=50",
             ),
         ]
     )
     def test_search(self, kwargs, params):
+        mock_request = self.mocker.patch("censys.asm.api.CensysAsmAPI.get_workspace_id")
+        mock_request.return_value = WORKSPACE_ID
+
         # Setup response
         self.responses.add(
             responses.GET,

tests/asm/utils.py

@@ -6,6 +6,7 @@
 V2_URL = f"{BASE_URL}/v2"
 BETA_URL = f"{BASE_URL}/beta"
 INVENTORY_URL = f"{BASE_URL}/inventory"
+WORKSPACE_ID = "test-workspace-id"
 
 
 class MockResponse: