Merge pull request #172 from censys/adh/asm-updates

ASM Updates

.github/workflows/python-ci.yml

@@ -5,8 +5,6 @@ on:
     paths-ignore:
       - "**.rst"
       - "**.md"
-    tags-ignore:
-      - "*"
 
 jobs:
   build:
@@ -28,15 +26,6 @@ jobs:
           python -m pip install --upgrade pip poetry
           poetry config virtualenvs.in-project true
 
-      - name: Cache poetry
-        id: cached-poetry-dependencies
-        uses: actions/cache@v2
-        with:
-          # This path is specific to Ubuntu
-          path: .venv
-          # Look to see if there is a cache hit for the corresponding requirements file
-          key: venv-${{ runner.os }}-py${{ matrix.python }}-${{ hashFiles('**/poetry.lock') }}
-
       - name: Install dependencies
         run: |
           poetry install
@@ -53,4 +42,4 @@ jobs:
 
       - name: Test with pytest
         run: |
-          poetry run pytest
+          poetry run pytest --cov --cov-fail-under=100

censys/asm/assets/assets.py

@@ -1,6 +1,6 @@
 """Base for interacting with the Censys Assets API."""
 import re
-from typing import Iterator, Optional
+from typing import Any, Dict, Iterator, List, Optional
 
 from ..api import CensysAsmAPI
 from censys.common.exceptions import CensysInvalidColorException
@@ -23,19 +23,34 @@ def __init__(self, asset_type: str, *args, **kwargs):
         self.base_path = f"assets/{asset_type}"
 
     def get_assets(
-        self, page_number: int = 1, page_size: Optional[int] = None
+        self,
+        page_number: int = 1,
+        page_size: Optional[int] = None,
+        tag: Optional[List[str]] = None,
+        tag_operator: Optional[str] = None,
+        source: Optional[List[str]] = None,
     ) -> Iterator[dict]:
         """Requests assets data.
 
         Args:
             page_number (int): Optional; Page number to begin at when searching.
             page_size (int): Optional; Page size for retrieving assets.
+            tag (list): Optional; List of tags to search for.
+            tag_operator (str): Optional; Operator to use when searching for tags.
+            source (list): Optional; List of sources to search for.
 
         Yields:
             dict: The assets result returned.
         """
+        args: Dict[str, Any] = {}
+        if tag:
+            args["tag"] = tag
+        if tag_operator:
+            args["tagOperator"] = tag_operator
+        if source:
+            args["source"] = source
         yield from self._get_page(
-            self.base_path, page_number=page_number, page_size=page_size
+            self.base_path, page_number=page_number, page_size=page_size, args=args
         )
 
     def get_asset_by_id(self, asset_id: str) -> dict:

censys/asm/seeds.py

@@ -11,18 +11,24 @@ class Seeds(CensysAsmAPI):
 
     base_path = "seeds"
 
-    def get_seeds(self, seed_type: Optional[str] = None) -> dict:
+    def get_seeds(
+        self, seed_type: Optional[str] = None, label: Optional[str] = None
+    ) -> dict:
         """Requests seed data.
 
         Args:
             seed_type (str):
                 Optional; Seed type ['IP_ADDRESS', 'DOMAIN_NAME', 'CIDR', 'ASN'].
+            label (str): Optional; Seed label.
 
         Returns:
             dict: Seed search results.
         """
-        args = {"type": seed_type}
-
+        args = {}
+        if seed_type:
+            args["type"] = seed_type
+        if label:
+            args["label"] = label
         return self._get(self.base_path, args=args)["seeds"]
 
     def get_seed_by_id(self, seed_id: int) -> dict:

censys/cli/commands/config.py

@@ -1,5 +1,6 @@
 """Censys config CLI."""
 import argparse
+import os
 import sys
 
 from rich.prompt import Prompt
@@ -22,6 +23,17 @@ def cli_config(_: argparse.Namespace):  # pragma: no cover
     api_id = config.get(DEFAULT, "api_id")
     api_secret = config.get(DEFAULT, "api_secret")
 
+    api_id_env = os.getenv("CENSYS_API_ID")
+    api_secret_env = os.getenv("CENSYS_API_SECRET")
+
+    if api_id_env is not None or api_secret_env is not None:
+        print(
+            "Please note environment variables (CENSYS_API_ID & CENSYS_API_SECRET) "
+            "will take priority over configured credentials."
+        )
+        api_id = api_id_env or api_id
+        api_secret = api_secret_env or api_secret
+
     if api_id and api_secret:
         redacted_id = api_id.replace(api_id[:32], 32 * "*")
         redacted_secret = api_secret.replace(api_secret[:28], 28 * "*")

censys/common/version.py

@@ -1,7 +1,7 @@
 """Censys Version."""
-try:
+try:  # pragma: no cover
     import importlib_metadata
-except ImportError:
+except ImportError:  # pragma: no cover
     import importlib.metadata as importlib_metadata  # type: ignore
 
 __version__: str = importlib_metadata.version("censys")

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "censys"
-version = "2.0.7"
+version = "2.0.8"
 description = "An easy-to-use and lightweight API wrapper for Censys APIs (censys.io)."
 readme = "README.md"
 authors = ["Censys, Inc. <[email protected]>"]

tests/asm/test_assets.py

@@ -58,6 +58,27 @@ def test_get_assets(self, mock):
             timeout=TEST_TIMEOUT,
         )
 
+    @patch("censys.common.base.requests.Session.get")
+    def test_get_assets_by_tag(self, mock):
+        mock.return_value = MockResponse(TEST_SUCCESS_CODE, ASSET_TYPE)
+        assets = getattr(self.client, self.asset_type).get_assets(
+            tag=[TEST_TAG_NAME], tag_operator="is", source=["Seed"]
+        )
+        res = list(assets)
+
+        assert RESOURCE_PAGING_RESULTS == res
+        mock.assert_called_with(
+            f"{ASSETS_URL}/{self.asset_type}",
+            params={
+                "pageNumber": 3,
+                "pageSize": 500,
+                "tag": [TEST_TAG_NAME],
+                "tagOperator": "is",
+                "source": ["Seed"],
+            },
+            timeout=TEST_TIMEOUT,
+        )
+
     @patch("censys.common.base.requests.Session.get")
     def test_get_hosts_by_page(self, mock):
         mock.return_value = MockResponse(

tests/asm/test_seeds.py

@@ -39,7 +39,18 @@ def test_get_seeds(self, mock):
         mock.return_value = MockResponse(TEST_SUCCESS_CODE, SEED_RESOURCE_TYPE)
         self.client.seeds.get_seeds()
 
-        mock.assert_called_with(SEEDS_URL, params={"type": None}, timeout=TEST_TIMEOUT)
+        mock.assert_called_with(SEEDS_URL, params={}, timeout=TEST_TIMEOUT)
+
+    @patch("censys.common.base.requests.Session.get")
+    def test_get_seeds_with_label(self, mock):
+        mock.return_value = MockResponse(TEST_SUCCESS_CODE, SEED_RESOURCE_TYPE)
+        self.client.seeds.get_seeds(seed_type="ASN", label=TEST_SEED_LABEL)
+
+        mock.assert_called_with(
+            SEEDS_URL,
+            params={"type": "ASN", "label": TEST_SEED_LABEL},
+            timeout=TEST_TIMEOUT,
+        )
 
     @patch("censys.common.base.requests.Session.get")
     def test_get_seeds_by_type(self, mock):