Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions .github/workflows/python-ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -37,10 +37,11 @@ jobs:
run: |
poetry install

- name: Lint with flake8
- name: Lint with ruff
if: startsWith(matrix.os, 'ubuntu')
run: |
poetry run flake8 --max-line-length=127
poetry run ruff check .
poetry run ruff format --check .

- name: Type checking with mypy
if: startsWith(matrix.os, 'ubuntu')
Expand Down
28 changes: 5 additions & 23 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -15,22 +15,18 @@ repos:
- id: end-of-file-fixer
- id: mixed-line-ending
- id: trailing-whitespace
- repo: https://github.com/asottile/pyupgrade
rev: v3.7.0
hooks:
- id: pyupgrade
args: [--py38-plus]
- repo: local
hooks:
- id: system
name: black
entry: poetry run black .
name: ruff
entry: poetry run ruff check . --fix
pass_filenames: false
language: system
types: [python]
require_serial: true
- id: system
name: isort
entry: poetry run isort .
name: ruff-format
entry: poetry run ruff format .
pass_filenames: false
language: system
types: [python]
Expand All @@ -40,24 +36,10 @@ repos:
pass_filenames: false
language: system
types: [python]
- id: system
name: flake8
entry: poetry run flake8 .
pass_filenames: false
language: system
types: [python]
require_serial: true
args: [--darglint-ignore-regex, .*] # Skip docstring checks
- id: system
name: mypy
entry: poetry run mypy censys
pass_filenames: false
language: system
types: [python]
require_serial: true
# - id: system
# name: update autocomplete
# entry: bash scripts/update_autocomplete.sh
# pass_filenames: false
# language: system
# types: []
23 changes: 7 additions & 16 deletions censys/asm/api.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,9 @@
"""Base for interacting with the Censys ASM API."""

import os
from collections.abc import Iterator
from math import inf
from typing import Iterator, Optional, Type
from typing import Optional

from requests.models import Response

Expand Down Expand Up @@ -38,25 +39,17 @@ def __init__(self, api_key: Optional[str] = None, **kwargs):
config = get_config()

# Try to get credentials
self._api_key = (
api_key
or os.getenv("CENSYS_ASM_API_KEY")
or config.get(DEFAULT, "asm_api_key")
)
self._api_key = api_key or os.getenv("CENSYS_ASM_API_KEY") or config.get(DEFAULT, "asm_api_key")

if not self._api_key:
raise CensysException("No ASM API key configured.")

self._session.headers.update(
{"Content-Type": "application/json", "Censys-Api-Key": self._api_key}
)
self._session.headers.update({"Content-Type": "application/json", "Censys-Api-Key": self._api_key})

def _get_exception_class( # type: ignore
self, res: Response
) -> Type[CensysAsmException]:
return CensysExceptionMapper.ASM_EXCEPTIONS.get(
res.json().get("errorCode"), CensysAsmException
)
) -> type[CensysAsmException]:
return CensysExceptionMapper.ASM_EXCEPTIONS.get(res.json().get("errorCode"), CensysAsmException)

def _get_page(
self,
Expand Down Expand Up @@ -91,9 +84,7 @@ def _get_page(

yield from res[keyword]

def _get_logbook_page(
self, path: str, args: Optional[dict] = None
) -> Iterator[dict]:
def _get_logbook_page(self, path: str, args: Optional[dict] = None) -> Iterator[dict]:
"""Fetches paginated ASM logbook API events.

Args:
Expand Down
28 changes: 11 additions & 17 deletions censys/asm/assets/assets.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
"""Base for interacting with the Censys Assets API."""

import re
from typing import Any, Dict, Iterator, List, Optional
from collections.abc import Iterator
from typing import Any, Optional

from ..api import CensysAsmAPI
from censys.common.exceptions import CensysInvalidColorException

from ..api import CensysAsmAPI

HEX_REGEX = re.compile(r"^#(?:[0-9a-fA-F]{3}){1,2}$")


Expand Down Expand Up @@ -42,9 +44,9 @@ def get_assets(
self,
page_number: int = 1,
page_size: Optional[int] = None,
tag: Optional[List[str]] = None,
tag: Optional[list[str]] = None,
tag_operator: Optional[str] = None,
source: Optional[List[str]] = None,
source: Optional[list[str]] = None,
discovery_trail: Optional[bool] = None,
) -> Iterator[dict]:
"""Requests assets data.
Expand All @@ -60,7 +62,7 @@ def get_assets(
Yields:
dict: The assets result returned.
"""
args: Dict[str, Any] = {}
args: dict[str, Any] = {}
if tag:
args["tag"] = tag
if tag_operator:
Expand All @@ -69,9 +71,7 @@ def get_assets(
args["source"] = source
if discovery_trail:
args["discoveryTrail"] = discovery_trail
yield from self._get_page(
self.base_path, page_number=page_number, page_size=page_size, args=args
)
yield from self._get_page(self.base_path, page_number=page_number, page_size=page_size, args=args)

def get_asset_by_id(self, asset_id: str) -> dict:
"""Requests asset data by ID.
Expand Down Expand Up @@ -104,9 +104,7 @@ def get_comments(
"""
path = f"{self.base_path}/{self._format_asset_id(asset_id)}/comments"

return self._get_page(
path, page_number=page_number, page_size=page_size, keyword="comments"
)
return self._get_page(path, page_number=page_number, page_size=page_size, keyword="comments")

def get_comment_by_id(self, asset_id: str, comment_id: int) -> dict:
"""Requests a comment on a specified asset by comment ID.
Expand All @@ -118,9 +116,7 @@ def get_comment_by_id(self, asset_id: str, comment_id: int) -> dict:
Returns:
dict: Comment search result.
"""
path = (
f"{self.base_path}/{self._format_asset_id(asset_id)}/comments/{comment_id}"
)
path = f"{self.base_path}/{self._format_asset_id(asset_id)}/comments/{comment_id}"

return self._get(path)

Expand Down Expand Up @@ -149,9 +145,7 @@ def delete_comment(self, asset_id: str, comment_id: int) -> dict:
Returns:
dict: Deleted comment results.
"""
path = (
f"{self.base_path}/{self._format_asset_id(asset_id)}/comments/{comment_id}"
)
path = f"{self.base_path}/{self._format_asset_id(asset_id)}/comments/{comment_id}"

return self._delete(path)

Expand Down
7 changes: 3 additions & 4 deletions censys/asm/assets/domains.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Interact with the Censys Domain Assets API."""

from typing import Iterator, Optional
from collections.abc import Iterator
from typing import Optional

from .assets import Assets

Expand All @@ -17,9 +18,7 @@ def __init__(self, *args, **kwargs):
"""
super().__init__("domains", *args, **kwargs)

def get_subdomains(
self, domain: str, page_number: int = 1, page_size: Optional[int] = None
) -> Iterator[dict]:
def get_subdomains(self, domain: str, page_number: int = 1, page_size: Optional[int] = None) -> Iterator[dict]:
"""List all subdomains of the parent domain.

Args:
Expand Down
9 changes: 5 additions & 4 deletions censys/asm/assets/subdomains.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Interact with the Censys Subdomain Assets API."""

from typing import Any, Dict, Iterator, List, Optional
from collections.abc import Iterator
from typing import Any, Optional

from .assets import Assets

Expand All @@ -21,9 +22,9 @@ def get_assets(
self,
page_number: int = 1,
page_size: Optional[int] = None,
tag: Optional[List[str]] = None,
tag: Optional[list[str]] = None,
tag_operator: Optional[str] = None,
source: Optional[List[str]] = None,
source: Optional[list[str]] = None,
discovery_trail: Optional[bool] = None,
) -> Iterator[dict]:
"""Requests assets data.
Expand All @@ -41,7 +42,7 @@ def get_assets(
Yields:
dict: The assets result returned.
"""
args: Dict[str, Any] = {}
args: dict[str, Any] = {}
if tag:
args["tag"] = tag
if tag_operator:
Expand Down
3 changes: 2 additions & 1 deletion censys/asm/assets/web_entities.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Interact with the Censys Web Entities Assets API."""

from typing import Iterator, Optional
from collections.abc import Iterator
from typing import Optional

from .assets import Assets

Expand Down
8 changes: 3 additions & 5 deletions censys/asm/beta.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
"""Interact with miscellaneous Censys Beta APIs."""

from typing import List, Optional
from typing import Optional

from ..common.types import Datetime
from ..common.utils import format_iso8601
Expand All @@ -12,9 +12,7 @@ class Beta(CensysAsmAPI):

base_path = "/beta"

def get_logbook_data(
self, filters: Optional[dict] = None, cursor: Optional[str] = None
):
def get_logbook_data(self, filters: Optional[dict] = None, cursor: Optional[str] = None):
"""Retrieve logbook data.

Args:
Expand All @@ -30,7 +28,7 @@ def get_logbook_data(
def add_cloud_assets(
self,
cloud_connector_uid: str,
cloud_assets: List[dict],
cloud_assets: list[dict],
):
"""Add cloud assets.

Expand Down
17 changes: 7 additions & 10 deletions censys/asm/inventory.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Interact with the Censys Inventory Search API."""

import warnings
from typing import List, Optional
from typing import Optional

from .api import CensysAsmAPI

Expand All @@ -13,12 +13,12 @@ class InventorySearch(CensysAsmAPI):

def search(
self,
workspaces: Optional[List[str]] = None,
workspaces: Optional[list[str]] = None,
query: Optional[str] = None,
page_size: Optional[int] = None,
cursor: Optional[str] = None,
sort: Optional[List[str]] = None,
fields: Optional[List[str]] = None,
sort: Optional[list[str]] = None,
fields: Optional[list[str]] = None,
pages: Optional[int] = None,
) -> dict:
"""Search inventory data.
Expand Down Expand Up @@ -75,10 +75,7 @@ def search(
while next_cursor and (pages == -1 or page < pages):
args["cursor"] = next_cursor
resp = self._get(self.base_path, args=args)
if "nextCursor" in resp:
next_cursor = resp.get("nextCursor")
else:
next_cursor = None
next_cursor = resp.get("nextCursor") if "nextCursor" in resp else None
hits.extend(resp.get("hits", []))
page += 1

Expand All @@ -87,7 +84,7 @@ def search(

def aggregate(
self,
workspaces: List[str],
workspaces: list[str],
query: Optional[str] = None,
aggregation: Optional[dict] = None,
) -> dict:
Expand All @@ -109,7 +106,7 @@ def aggregate(

return self._post(f"{self.base_path}/aggregate", data=body)

def fields(self, fields: Optional[List[str]] = None) -> dict:
def fields(self, fields: Optional[list[str]] = None) -> dict:
"""List inventory fields.

If no fields are specified, all fields will be returned.
Expand Down
7 changes: 4 additions & 3 deletions censys/asm/logbook.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
"""Interact with the Censys Logbook API."""

import datetime
from typing import Iterator, List, Optional, Union
from collections.abc import Iterator
from typing import Optional, Union

from .api import CensysAsmAPI

Expand All @@ -14,7 +15,7 @@ class Logbook(CensysAsmAPI):
def get_cursor(
self,
start: Optional[Union[datetime.datetime, int]] = None,
filters: Optional[List[str]] = None,
filters: Optional[list[str]] = None,
) -> str:
"""Requests a logbook cursor.

Expand Down Expand Up @@ -71,7 +72,7 @@ class Filters:

def format_data(
start: Optional[Union[datetime.datetime, int]] = None,
filters: Optional[List[str]] = None,
filters: Optional[list[str]] = None,
) -> dict:
"""Formats cursor request data into a start date/id and filter list.

Expand Down
Loading
Loading