Manager

psengine.asi.asi_mgr.AttackSurfaceMgr

AttackSurfaceMgr(api_token: str = None)

Manages requests for Recorded Future SecurityTrails (ASI) API.

PARAMETER	DESCRIPTION
api_token	ASI API token. TYPE: str DEFAULT: None

Source code in psengine/asi/asi_mgr.py

def __init__(self, api_token: str = None):
    """Initializes the `AttackSurfaceMgr` object.

    Args:
        api_token (str, optional): ASI API token.
    """
    self.log = logging.getLogger(__name__)
    self.asi_client = ASIClient(api_token=api_token) if api_token else ASIClient()

fetch_projects

fetch_projects(
    sort_direction: Optional[
        Literal['asc', 'desc']
    ] = 'asc',
) -> ProjectListOut

Fetch ASI projects.

PARAMETER	DESCRIPTION
sort_direction	Sort direction for the projects TYPE: Optional[Literal['asc', 'desc']] DEFAULT: 'asc'

Endpoint

v2/projects

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
ASIFetchProjectsError	If an API or connection error occurs.

RETURNS	DESCRIPTION
ProjectListOut	List of ASI Project models

Source code in psengine/asi/asi_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ASIFetchProjectsError)
def fetch_projects(
    self,
    sort_direction: Annotated[
        Optional[Literal['asc', 'desc']], Doc('Sort direction for the projects')
    ] = 'asc',
) -> Annotated[ProjectListOut, Doc('List of ASI Project models')]:
    """Fetch ASI projects.

    Endpoint:
        `v2/projects`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        ASIFetchProjectsError: If an API or connection error occurs.
    """
    params = {}
    if sort_direction:
        params['sort_direction'] = sort_direction

    response = self.asi_client.request('get', EP_ASI_PROJECTS, params=params).json()
    return ProjectListOut.model_validate({'data': response['data'], 'meta': response['meta']})

search_assets

search_assets(
    project_id: str,
    quick_search: Optional[str] = None,
    asset_id: Optional[str] = None,
    asset_name: Optional[str] = None,
    asset_apex_domain: Optional[
        Union[str, list[str]]
    ] = None,
    asset_discovered_date: Optional[
        tuple[Optional[str], Optional[str]]
    ] = None,
    asset_type: Optional[AssetType] = None,
    custom_tags: Optional[list[str]] = None,
    is_static_asset: Optional[bool] = None,
    exposure_severity: Optional[
        Union[ExposureSeverity, list[ExposureSeverity]]
    ] = None,
    exposure_signature_id: Optional[
        Union[str, list[str]]
    ] = None,
    exposure_score: Optional[tuple[int, int]] = None,
    exposure_last_scanned: Optional[
        tuple[Optional[str], Optional[str]]
    ] = None,
    open_port_number: Optional[
        Union[int, list[int]]
    ] = None,
    open_port_service: Optional[
        Union[str, list[str]]
    ] = None,
    open_port_protocol: Optional[
        Union[str, list[str]]
    ] = None,
    technology_name: Optional[Union[str, list[str]]] = None,
    certificate_issuer: Optional[
        Union[str, list[str]]
    ] = None,
    is_responsive: Optional[bool] = None,
    enrichments: list[EnrichmentType] = None,
    sort_by: list[SortByType] = None,
    assets_per_page: int = ASSETS_PER_PAGE,
    max_results: Optional[int] = DEFAULT_LIMIT,
) -> AssetResponse

Search for assets within an ASI project.

Does pagination requests on batches of assets_per_page up to max_results.

PARAMETER	DESCRIPTION
project_id	The ID of the ASI project to search assets within TYPE: str
quick_search	Search term to match against asset name, IP addresses, and technology fields TYPE: Optional[str] DEFAULT: None
asset_id	Filter for the specific asset, which will be either a IP or domain value (examples: 192.88.99.2 or www.example.com). TYPE: Optional[str] DEFAULT: None
asset_name	Filter on the name of the asset(IP address or domain). TYPE: Optional[str] DEFAULT: None
asset_apex_domain	Filter on the apex domain of the asset (example: example.com). Pass a single value or a list. TYPE: Optional[Union[str, list[str]]] DEFAULT: None
asset_discovered_date	Filter on the date (Y-m-d) the asset was discovered by Recorded Future ASI. This may be different than when the asset was added to the project. IPv4 addresses will have a fixed point in the past for their discovery date. Use None for an open-ended bound. TYPE: Optional[tuple[Optional[str], Optional[str]]] DEFAULT: None
asset_type	The type of asset, one of: ip, domain and host (where domain and host represent the same asset type). TYPE: Optional[AssetType] DEFAULT: None
custom_tags	Filter for assets tagged with any of the provided custom tags. TYPE: Optional[list[str]] DEFAULT: None
is_static_asset	Filter for assets that are static, meaning they have a consistent IP address or domain name over time. TYPE: Optional[bool] DEFAULT: None
exposure_severity	Filter assets by exposure severity. Pass a single value or a list to match any of the provided severities. TYPE: Optional[Union[ExposureSeverity, list[ExposureSeverity]]] DEFAULT: None
exposure_signature_id	Filter assets by ASI Signature ID. Pass a single ID or a list. Some signatures align with CVEs, e.g. "cve-2024-6387" or "cve-OpenSSH". TYPE: Optional[Union[str, list[str]]] DEFAULT: None
exposure_score	Filter assets by exposure score range (0–100). Provide a (min, max) tuple. The score indicates potential asset risk based on various factors. TYPE: Optional[tuple[int, int]] DEFAULT: None
exposure_last_scanned	Filter assets by the date they were last scanned for exposures. Provide a (start, end) tuple of "YYYY-MM-DD" strings. Use None for an open-ended bound. TYPE: Optional[tuple[Optional[str], Optional[str]]] DEFAULT: None
open_port_number	Filter for assets which have an open port with the provided number (e.g. 80). TYPE: Optional[Union[int, list[int]]] DEFAULT: None
open_port_service	Filter for assets which have an open port with the provided service (e.g. http, ftp, rdp). TYPE: Optional[Union[str, list[str]]] DEFAULT: None
open_port_protocol	Filter for assets which have an open port with the provided protocol (e.g. tcp, udp). TYPE: Optional[Union[str, list[str]]] DEFAULT: None
technology_name	Filter for the name of a technology found on the asset. Could be directly attached to the port (nginx, etc) or a web technology (e.g. 'jQuery', 'Wordpress')). TYPE: Optional[Union[str, list[str]]] DEFAULT: None
certificate_issuer	Filter where the certificate (or in the chain) issuer's common name or organization matches the provided value TYPE: Optional[Union[str, list[str]]] DEFAULT: None
is_responsive	Filter for assets that are unresponsive over ICMP and no ports are open. This is a boolean filter, so it will return assets that are either responsive or not responsive. TYPE: Optional[bool] DEFAULT: None
enrichments	List of enrichments to apply to the assets TYPE: list[EnrichmentType] DEFAULT: None
sort_by	List of fields to sort the assets by TYPE: list[SortByType] DEFAULT: None
assets_per_page	Number of assets to fetch per page TYPE: int DEFAULT: ASSETS_PER_PAGE
max_results	Maximum number of assets to fetch TYPE: Optional[int] DEFAULT: DEFAULT_LIMIT

Endpoint

v2/projects/{project_id}/assets/_search

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
ValueError	If exposure_score start is greater than end.
ASISearchAssetsError	If an API or connection error occurs.

RETURNS	DESCRIPTION
AssetResponse	Response model for ASI assets search

Source code in psengine/asi/asi_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ASISearchAssetsError)
def search_assets(
    self,
    project_id: Annotated[str, Doc('The ID of the ASI project to search assets within')],
    quick_search: Annotated[
        Optional[str],
        Doc('Search term to match against asset name, IP addresses, and technology fields'),
    ] = None,
    asset_id: Annotated[
        Optional[str],
        Doc(
            """Filter for the specific asset, which will be either a IP or domain value
            (examples: 192.88.99.2 or www.example.com)."""
        ),
    ] = None,
    asset_name: Annotated[
        Optional[str], Doc("""Filter on the name of the asset(IP address or domain).""")
    ] = None,
    asset_apex_domain: Annotated[
        Optional[Union[str, list[str]]],
        Doc(
            """Filter on the apex domain of the asset (example: example.com).
            Pass a single value or a list."""
        ),
    ] = None,
    asset_discovered_date: Annotated[
        Optional[tuple[Optional[str], Optional[str]]],
        Doc(
            """Filter on the date (Y-m-d) the asset was discovered by Recorded Future ASI.
            This may be different than when the asset was added to the project.
            IPv4 addresses will have a fixed point in the past for their discovery date.
            Use None for an open-ended bound."""
        ),
    ] = None,
    asset_type: Annotated[
        Optional[AssetType],
        Doc(
            """The type of asset, one of: ip, domain and host
            (where domain and host represent the same asset type)."""
        ),
    ] = None,
    custom_tags: Annotated[
        Optional[list[str]],
        Doc('Filter for assets tagged with any of the provided custom tags.'),
    ] = None,
    is_static_asset: Annotated[
        Optional[bool],
        Doc(
            """Filter for assets that are static, meaning they have a consistent IP address or
            domain name over time."""
        ),
    ] = None,
    exposure_severity: Annotated[
        Optional[Union[ExposureSeverity, list[ExposureSeverity]]],
        Doc("""Filter assets by exposure severity.
        Pass a single value or a list to match any of the provided severities."""),
    ] = None,
    exposure_signature_id: Annotated[
        Optional[Union[str, list[str]]],
        Doc("""Filter assets by ASI Signature ID. Pass a single ID or a list.
        Some signatures align with CVEs, e.g. "cve-2024-6387" or "cve-OpenSSH"."""),
    ] = None,
    exposure_score: Annotated[
        Optional[
            Annotated[
                tuple[Annotated[int, Field(ge=0, le=100)], Annotated[int, Field(ge=0, le=100)]],
                AfterValidator(_validate_exposure_score_range),
            ]
        ],
        Doc("""Filter assets by exposure score range (0–100). Provide a (min, max) tuple.
        The score indicates potential asset risk based on various factors."""),
    ] = None,
    exposure_last_scanned: Annotated[
        Optional[tuple[Optional[str], Optional[str]]],
        Doc("""Filter assets by the date they were last scanned for exposures.
        Provide a (start, end) tuple of "YYYY-MM-DD" strings.
        Use None for an open-ended bound."""),
    ] = None,
    open_port_number: Annotated[
        Optional[Union[int, list[int]]],
        Doc(
            """Filter for assets which have an open port with the provided number (e.g. 80)."""
        ),
    ] = None,
    open_port_service: Annotated[
        Optional[Union[str, list[str]]],
        Doc(
            """Filter for assets which have an open port with the provided service (e.g. http,
            ftp, rdp)."""
        ),
    ] = None,
    open_port_protocol: Annotated[
        Optional[Union[str, list[str]]],
        Doc(
            """Filter for assets which have an open port with the provided protocol (e.g. tcp,
            udp)."""
        ),
    ] = None,
    technology_name: Annotated[
        Optional[Union[str, list[str]]],
        Doc(
            """Filter for the name of a technology found on the asset. Could be directly
            attached to the port (nginx, etc) or a web technology (e.g. 'jQuery',
            'Wordpress'))."""
        ),
    ] = None,
    certificate_issuer: Annotated[
        Optional[Union[str, list[str]]],
        Doc(
            """Filter where the certificate (or in the chain) issuer's common name
            or organization matches the provided value"""
        ),
    ] = None,
    is_responsive: Annotated[
        Optional[bool],
        Doc(
            """Filter for assets that are unresponsive over ICMP and no ports are open.
            This is a boolean filter, so it will return assets that are either responsive
            or not responsive."""
        ),
    ] = None,
    enrichments: Annotated[
        list[EnrichmentType], Doc('List of enrichments to apply to the assets')
    ] = None,
    sort_by: Annotated[list[SortByType], Doc('List of fields to sort the assets by')] = None,
    assets_per_page: Annotated[
        int,
        Field(ge=1, le=MAX_ASI_PAGE_SIZE),
        Doc('Number of assets to fetch per page'),
    ] = ASSETS_PER_PAGE,
    max_results: Annotated[
        Optional[int], Doc('Maximum number of assets to fetch')
    ] = DEFAULT_LIMIT,
) -> Annotated[AssetResponse, Doc('Response model for ASI assets search')]:
    """Search for assets within an ASI project.

    Does pagination requests on batches of `assets_per_page` up to `max_results`.

    Endpoint:
        `v2/projects/{project_id}/assets/_search`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        ValueError: If `exposure_score` start is greater than end.
        ASISearchAssetsError: If an API or connection error occurs.
    """
    if sort_by is None:
        sort_by = ['discovered_at']
    filter_params = locals()
    for param in [
        'self',
        'project_id',
        'enrichments',
        'sort_by',
        'assets_per_page',
        'max_results',
    ]:
        filter_params.pop(param)

    filter_dict = self._lookup_filter(**filter_params)
    body = {'pagination': {'limit': assets_per_page}}
    if filter_dict:
        body['filter'] = filter_dict

    if enrichments:
        body['enrichments'] = enrichments
    if sort_by:
        body['sort'] = sort_by

    data = AssetSearchRequest.model_validate(body).model_dump(
        by_alias=True, exclude_none=True, mode='json'
    )
    response = self.asi_client.request_paged(
        'post', EP_ASI_ASSETS_SEARCH.format(project_id), data=data, max_results=max_results
    )

    return AssetResponse.model_validate({'data': response['data'], 'meta': response['meta']})

search_exposures

search_exposures(
    project_id: str,
    filter_cve_id: Optional[str] = None,
    filter_cvss_score_gte: Optional[float] = None,
    filter_cvss_score_lte: Optional[float] = None,
    filter_cwe_id: Optional[str] = None,
    filter_severity_exact: Optional[SEVERITY_FILTER] = None,
    filter_severity_min: Optional[SEVERITY_FILTER] = None,
    max_results: Optional[int] = DEFAULT_LIMIT,
    exposures_per_page: Optional[int] = DEFAULT_LIMIT,
) -> ExposureSearchOut

Search for exposures within an ASI project.

Does pagination requests on batches of the API default page size up to max_results.

PARAMETER	DESCRIPTION
project_id	The ID of the ASI project to search assets within TYPE: str
filter_cve_id	Filter for asset or exposure tied to a vulnerability with the provided CVE. Example CVE-2024-6387. TYPE: Optional[str] DEFAULT: None
filter_cvss_score_gte	Filter for asset or exposure tied to a vulnerability with the provided CVSS score range. Example 7.5. TYPE: Optional[float] DEFAULT: None
filter_cvss_score_lte	Filter for asset or exposure tied to a vulnerability with the provided CVSS score range. Example 7.5. TYPE: Optional[float] DEFAULT: None
filter_cwe_id	Filter for asset or exposure tied to a vulnerability associated with the provided CWE. Example CWE-79. TYPE: Optional[str] DEFAULT: None
filter_severity_exact	Filter for assets which have an exposure severity matching the provided value. TYPE: Optional[SEVERITY_FILTER] DEFAULT: None
filter_severity_min	Filter for assets which have an exposure severity matching or higher than the provided value. TYPE: Optional[SEVERITY_FILTER] DEFAULT: None
max_results	Maximum number of assets to fetch TYPE: Optional[int] DEFAULT: DEFAULT_LIMIT
exposures_per_page	Results per page TYPE: Optional[int] DEFAULT: DEFAULT_LIMIT

Endpoint

v2/projects/{project_id}/exposures

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
ASIExposureSearchError	If an API or connection error occurs.

RETURNS	DESCRIPTION
ExposureSearchOut	Response model for ASI exposures search

Source code in psengine/asi/asi_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ASIExposureSearchError)
def search_exposures(
    self,
    project_id: Annotated[str, Doc('The ID of the ASI project to search assets within')],
    filter_cve_id: Annotated[
        Optional[str],
        Doc(
            """Filter for asset or exposure tied to a vulnerability with the provided CVE.
            Example CVE-2024-6387."""
        ),
    ] = None,
    filter_cvss_score_gte: Annotated[
        Optional[float],
        Field(ge=0, le=10),
        Doc(
            """Filter for asset or exposure tied to a vulnerability with the provided CVSS
            score range. Example 7.5. """
        ),
    ] = None,
    filter_cvss_score_lte: Annotated[
        Optional[float],
        Field(ge=0, le=10),
        Doc(
            """Filter for asset or exposure tied to a vulnerability with the provided CVSS
            score range. Example 7.5."""
        ),
    ] = None,
    filter_cwe_id: Annotated[
        Optional[str],
        Doc(
            """Filter for asset or exposure tied to a vulnerability associated with
            the provided CWE. Example CWE-79."""
        ),
    ] = None,
    filter_severity_exact: Annotated[
        Optional[SEVERITY_FILTER],
        Doc('Filter for assets which have an exposure severity matching the provided value.'),
    ] = None,
    filter_severity_min: Annotated[
        Optional[SEVERITY_FILTER],
        Doc(
            """Filter for assets which have an exposure severity matching or higher than the
            provided value."""
        ),
    ] = None,
    max_results: Annotated[
        Optional[int], Doc('Maximum number of assets to fetch')
    ] = DEFAULT_LIMIT,
    exposures_per_page: Annotated[Optional[int], Doc('Results per page')] = DEFAULT_LIMIT,
) -> Annotated[ExposureSearchOut, Doc('Response model for ASI exposures search')]:
    """Search for exposures within an ASI project.

    Does pagination requests on batches of the API default page size up to `max_results`.

    Endpoint:
        `v2/projects/{project_id}/exposures`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        ASIExposureSearchError: If an API or connection error occurs.
    """
    params = {
        k: v
        for k, v in locals().items()
        if k not in ('self', 'max_results', 'exposures_per_page', 'project_id')
    }
    data = self.asi_client.request_paged(
        'GET',
        EP_ASI_EXPOSURES.format(project_id),
        params=params,
        max_results=max_results,
        objects_per_page=exposures_per_page,
    )

    return ExposureSearchOut.model_validate({'data': data['data'], 'meta': data['meta']})

fetch_exposures_by_signature

fetch_exposures_by_signature(
    project_id: str,
    signature_id: str,
    max_results: Optional[int] = DEFAULT_LIMIT,
    exposures_per_page: Optional[int] = DEFAULT_LIMIT,
) -> AssetWithExposureSearch

Fetch assets by exposure signature within an ASI project.

Does pagination requests on batches of the API default page size up to max_results.

PARAMETER	DESCRIPTION
project_id	The ID of the ASI project to search assets within TYPE: str
signature_id	The ID of the signature to search assets within TYPE: str
max_results	Maximum number of assets to fetch TYPE: Optional[int] DEFAULT: DEFAULT_LIMIT
exposures_per_page	Results per page TYPE: Optional[int] DEFAULT: DEFAULT_LIMIT

Endpoint

v2/projects/{project_id}/exposures/{signature_id}

RAISES	DESCRIPTION
ValidationError	If any supplied parameter is of incorrect type.
ASIFetchExposureError	If an API or connection error occurs.

RETURNS	DESCRIPTION
AssetWithExposureSearch	ASI asset with exposure details for the requested signature

Source code in psengine/asi/asi_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ASIFetchExposureError)
def fetch_exposures_by_signature(
    self,
    project_id: Annotated[str, Doc('The ID of the ASI project to search assets within')],
    signature_id: Annotated[str, Doc('The ID of the signature to search assets within')],
    max_results: Annotated[
        Optional[int], Doc('Maximum number of assets to fetch')
    ] = DEFAULT_LIMIT,
    exposures_per_page: Annotated[Optional[int], Doc('Results per page')] = DEFAULT_LIMIT,
) -> Annotated[
    AssetWithExposureSearch, Doc('ASI asset with exposure details for the requested signature')
]:
    """Fetch assets by exposure signature within an ASI project.

    Does pagination requests on batches of the API default page size up to `max_results`.

    Endpoint:
        `v2/projects/{project_id}/exposures/{signature_id}`

    Raises:
        ValidationError: If any supplied parameter is of incorrect type.
        ASIFetchExposureError: If an API or connection error occurs.
    """
    params = {
        k: v
        for k, v in locals().items()
        if k not in ('self', 'max_results', 'exposures_per_page', 'project_id', 'signature_id')
    }
    data = self.asi_client.request_paged(
        'GET',
        EP_ASI_EXPOSURES_BY_SIGNATURE.format(project_id, signature_id),
        params=params,
        max_results=max_results,
        objects_per_page=exposures_per_page,
    )
    return AssetWithExposureSearch.model_validate(data['data'][0])

fetch_assets

fetch_assets(
    project_id: str,
    sort_by: Literal[
        'discovered_at',
        'added_to_project_at',
        'last_scanned_at',
        'exposure_score',
        'asset_id',
        'apex_domain',
    ] = 'exposure_score',
    sort_direction: Literal['asc', 'desc'] = 'desc',
    asset_type: Optional[
        Literal['domain', 'host', 'ip']
    ] = None,
    custom_tags: Optional[str] = None,
    custom_tags_strict: Optional[str] = None,
    has_custom_tags: Optional[bool] = None,
    added_to_project_before: Optional[str] = None,
    added_to_project_after: Optional[str] = None,
    discovered_before: Optional[str] = None,
    discovered_after: Optional[str] = None,
    apex: Optional[str] = None,
    referenced_ip: Optional[str] = None,
    referenced_ip_before: Optional[str] = None,
    referenced_ip_after: Optional[str] = None,
    has_dns_record_type: Optional[str] = None,
    dns_resolves: Optional[bool] = None,
    asn: Optional[int] = None,
    cname_reference: Optional[str] = None,
    geo_country_iso: Optional[str] = None,
    ip_owner: Optional[str] = None,
    whois_email: Optional[str] = None,
    whois_email_current: Optional[str] = None,
    open_port_number: Optional[int] = None,
    open_port_protocol: Optional[str] = None,
    open_port_service: Optional[str] = None,
    open_port_technology: Optional[str] = None,
    technology_name: Optional[str] = None,
    web_technology_name: Optional[str] = None,
    certificate_issuer: Optional[str] = None,
    certificate_expires_before: Optional[str] = None,
    certificate_expires_after: Optional[str] = None,
    certificate_issued_before: Optional[str] = None,
    certificate_issued_after: Optional[str] = None,
    certificate_subject: Optional[str] = None,
    certificate_subject_alt_name: Optional[str] = None,
    certificate_sha256: Optional[str] = None,
    certificate_covers_domain: Optional[str] = None,
    waf_detected: Optional[bool] = None,
    waf_name: Optional[str] = None,
    is_responsive: Optional[bool] = None,
    exposure_score_gte: Optional[int] = None,
    exposure_score_lte: Optional[int] = None,
    exposure_severity: Optional[
        Literal[
            'unknown',
            'informational',
            'moderate',
            'critical',
        ]
    ] = None,
    exposure_id: Optional[str] = None,
    additional_fields: Optional[
        list[
            Literal[
                'custom_tags',
                'dns_records',
                'whois',
                'ip_metadata',
                'open_tcp_ports',
                'open_udp_ports',
                'web_technologies',
                'certificates',
                'certificate_chain',
                'defenses',
                'exposures',
                'exposure_instance_details',
            ]
        ]
    ] = None,
    max_results: Optional[int] = DEFAULT_LIMIT,
    assets_per_page: Optional[int] = DEFAULT_LIMIT,
) -> AssetResponse

Fetch assets within an ASI project.

Does pagination requests on batches of the API default page size up to max_results.

PARAMETER	DESCRIPTION
project_id	The ID of the ASI project to search assets within TYPE: str
sort_by	The field to sort by. TYPE: Literal['discovered_at', 'added_to_project_at', 'last_scanned_at', 'exposure_score', 'asset_id', 'apex_domain'] DEFAULT: 'exposure_score'
sort_direction	The direction to sort by. TYPE: Literal['asc', 'desc'] DEFAULT: 'desc'
asset_type	The type of asset, one of: ip, domain, or host. TYPE: Optional[Literal['domain', 'host', 'ip']] DEFAULT: None
custom_tags	Filter by custom tags placed on your assets. TYPE: Optional[str] DEFAULT: None
custom_tags_strict	Filter by custom tags placed on your assets. Strict version will return a validation error if any of the tags have not been defined on your project. TYPE: Optional[str] DEFAULT: None
has_custom_tags	Filter for assets that have at least one custom tag applied. Overrides any other custom tag filtering specified. TYPE: Optional[bool] DEFAULT: None
added_to_project_before	Filter on the date (YYYY-MM-DD) the asset was added to the project. TYPE: Optional[str] DEFAULT: None
added_to_project_after	Filter on the date (YYYY-MM-DD) the asset was added to the project. TYPE: Optional[str] DEFAULT: None
discovered_before	Filter on the date (YYYY-MM-DD) the asset was discovered. TYPE: Optional[str] DEFAULT: None
discovered_after	Filter on the date (YYYY-MM-DD) the asset was discovered. TYPE: Optional[str] DEFAULT: None
apex	Filter on the apex domain of the assets. Example: example.com. TYPE: Optional[str] DEFAULT: None
referenced_ip	Filter on an A or CNAME record pointing to the IP address. Use eq or in for exact IP matching. Use contains with a trailing . for CIDR range matching, or without for prefix matching. TYPE: Optional[str] DEFAULT: None
referenced_ip_before	If filtering on a referenced_ip, include additional criteria that the record existed during a date range. The reference must have started before this date. TYPE: Optional[str] DEFAULT: None
referenced_ip_after	If filtering on a referenced_ip, include additional criteria that the record existed during a date range. The reference must have existed after this date. TYPE: Optional[str] DEFAULT: None
has_dns_record_type	Filter for assets that have this DNS record type, e.g. A, CNAME, MX. TYPE: Optional[str] DEFAULT: None
dns_resolves	Filter for assets that in the end resolve to a valid IP currently, either via an A or CNAME. IP assets are included when filtering for assets that resolve. TYPE: Optional[bool] DEFAULT: None
asn	Filter for assets which either are, or point to, an IP address announced by the provided ASN. TYPE: Optional[int] DEFAULT: None
cname_reference	Filter on a domain that is referenced by a CNAME record. Only makes sense for domain asset types. Treated as a wildcard. TYPE: Optional[str] DEFAULT: None
geo_country_iso	Filter for assets which either are, or point to, an IP address located in the provided ISO country code. TYPE: Optional[str] DEFAULT: None
ip_owner	Filter for assets which either are, or point to, an IP address owned by the provided organization. TYPE: Optional[str] DEFAULT: None
whois_email	Filter for assets where the WHOIS email address matches the provided value. TYPE: Optional[str] DEFAULT: None
whois_email_current	Filter for assets where the WHOIS email address matches the provided value on the current WHOIS record. TYPE: Optional[str] DEFAULT: None
open_port_number	Filter for assets which have an open port with the provided number. TYPE: Optional[int] DEFAULT: None
open_port_protocol	Filter for assets which have an open port on the provided protocol. TYPE: Optional[str] DEFAULT: None
open_port_service	Filter for assets which have an open port that appears to support the provided protocol. TYPE: Optional[str] DEFAULT: None
open_port_technology	Filter for assets which have a specific product listening on an open port. TYPE: Optional[str] DEFAULT: None
technology_name	Filter for the name of a technology found on the asset. TYPE: Optional[str] DEFAULT: None
web_technology_name	Filter for the name of a technology specifically associated with web resources, such as jQuery or Wordpress. TYPE: Optional[str] DEFAULT: None
certificate_issuer	Filter where the certificate issuer's common name or organization matches the provided value. TYPE: Optional[str] DEFAULT: None
certificate_expires_before	Filter where the certificate expiration date is before the provided value. TYPE: Optional[str] DEFAULT: None
certificate_expires_after	Filter where the certificate expiration date is after the provided value. TYPE: Optional[str] DEFAULT: None
certificate_issued_before	Filter where the certificate issuance date is before the provided value. TYPE: Optional[str] DEFAULT: None
certificate_issued_after	Filter where the certificate issuance date is after the provided value. TYPE: Optional[str] DEFAULT: None
certificate_subject	Filter where certificate subject or organizationName matches the value. TYPE: Optional[str] DEFAULT: None
certificate_subject_alt_name	Filter where the certificate Subject Alternative Name matches the value. TYPE: Optional[str] DEFAULT: None
certificate_sha256	Filter where the certificate public key sha256 value matches the value. TYPE: Optional[str] DEFAULT: None
certificate_covers_domain	Filter where the certificate subject common name or SAN exactly matches or wildcard-covers the provided value. TYPE: Optional[str] DEFAULT: None
waf_detected	Filter for assets where a WAF is detected. TYPE: Optional[bool] DEFAULT: None
waf_name	Filter for assets where a specific WAF is detected. TYPE: Optional[str] DEFAULT: None
is_responsive	Filter for assets that are either responsive or not responsive over ICMP and port scanning. TYPE: Optional[bool] DEFAULT: None
exposure_score_gte	Filter for assets with exposure score greater than or equal to this value. TYPE: Optional[int] DEFAULT: None
exposure_score_lte	Filter for assets with exposure score less than or equal to this value. TYPE: Optional[int] DEFAULT: None
exposure_severity	Filter for assets with an exposure severity matching or higher than the provided value. TYPE: Optional[Literal['unknown', 'informational', 'moderate', 'critical']] DEFAULT: None
exposure_id	Filter for assets which have an exposure with the provided ASI Signature ID. TYPE: Optional[str] DEFAULT: None
additional_fields	Additional fields to include in the response. May be specified multiple times or as a comma-separated list in the raw API. TYPE: Optional[list[Literal['custom_tags', 'dns_records', 'whois', 'ip_metadata', 'open_tcp_ports', 'open_udp_ports', 'web_technologies', 'certificates', 'certificate_chain', 'defenses', 'exposures', 'exposure_instance_details']]] DEFAULT: None
max_results	Maximum number of assets to fetch TYPE: Optional[int] DEFAULT: DEFAULT_LIMIT
assets_per_page	Results per page TYPE: Optional[int] DEFAULT: DEFAULT_LIMIT

Endpoint

v2/projects/{project_id}/assets

RAISES	DESCRIPTION
ValidationError	If response data does not match the AssetResponse model.
ASIFetchAssetError	If an API or connection error occurs.

RETURNS	DESCRIPTION
AssetResponse	Response model for ASI assets list

Source code in psengine/asi/asi_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ASIFetchAssetError)
def fetch_assets(
    self,
    project_id: Annotated[str, Doc('The ID of the ASI project to search assets within')],
    sort_by: Annotated[
        Literal[
            'discovered_at',
            'added_to_project_at',
            'last_scanned_at',
            'exposure_score',
            'asset_id',
            'apex_domain',
        ],
        Doc('The field to sort by.'),
    ] = 'exposure_score',
    sort_direction: Annotated[
        Literal['asc', 'desc'],
        Doc('The direction to sort by.'),
    ] = 'desc',
    asset_type: Annotated[
        Optional[Literal['domain', 'host', 'ip']],
        Doc('The type of asset, one of: ip, domain, or host.'),
    ] = None,
    custom_tags: Annotated[
        Optional[str],
        Doc('Filter by custom tags placed on your assets.'),
    ] = None,
    custom_tags_strict: Annotated[
        Optional[str],
        Doc(
            'Filter by custom tags placed on your assets. Strict version will return a '
            'validation error if any of the tags have not been defined on your project.'
        ),
    ] = None,
    has_custom_tags: Annotated[
        Optional[bool],
        Doc(
            'Filter for assets that have at least one custom tag applied. Overrides any '
            'other custom tag filtering specified.'
        ),
    ] = None,
    added_to_project_before: Annotated[
        Optional[str],
        Doc('Filter on the date (YYYY-MM-DD) the asset was added to the project.'),
    ] = None,
    added_to_project_after: Annotated[
        Optional[str],
        Doc('Filter on the date (YYYY-MM-DD) the asset was added to the project.'),
    ] = None,
    discovered_before: Annotated[
        Optional[str],
        Doc('Filter on the date (YYYY-MM-DD) the asset was discovered.'),
    ] = None,
    discovered_after: Annotated[
        Optional[str],
        Doc('Filter on the date (YYYY-MM-DD) the asset was discovered.'),
    ] = None,
    apex: Annotated[
        Optional[str],
        Doc('Filter on the apex domain of the assets. Example: example.com.'),
    ] = None,
    referenced_ip: Annotated[
        Optional[str],
        Doc(
            'Filter on an A or CNAME record pointing to the IP address. Use eq or in for '
            'exact IP matching. Use contains with a trailing . for CIDR range matching, '
            'or without for prefix matching.'
        ),
    ] = None,
    referenced_ip_before: Annotated[
        Optional[str],
        Doc(
            'If filtering on a referenced_ip, include additional criteria that the record '
            'existed during a date range. The reference must have started before this date.'
        ),
    ] = None,
    referenced_ip_after: Annotated[
        Optional[str],
        Doc(
            'If filtering on a referenced_ip, include additional criteria that the record '
            'existed during a date range. The reference must have existed after this date.'
        ),
    ] = None,
    has_dns_record_type: Annotated[
        Optional[str],
        Doc('Filter for assets that have this DNS record type, e.g. A, CNAME, MX.'),
    ] = None,
    dns_resolves: Annotated[
        Optional[bool],
        Doc(
            'Filter for assets that in the end resolve to a valid IP currently, either via '
            'an A or CNAME. IP assets are included when filtering for assets that resolve.'
        ),
    ] = None,
    asn: Annotated[
        Optional[int],
        Doc(
            'Filter for assets which either are, or point to, an IP address announced by '
            'the provided ASN.'
        ),
    ] = None,
    cname_reference: Annotated[
        Optional[str],
        Doc(
            'Filter on a domain that is referenced by a CNAME record. Only makes sense for '
            'domain asset types. Treated as a wildcard.'
        ),
    ] = None,
    geo_country_iso: Annotated[
        Optional[str],
        Doc(
            'Filter for assets which either are, or point to, an IP address located in the '
            'provided ISO country code.'
        ),
    ] = None,
    ip_owner: Annotated[
        Optional[str],
        Doc(
            'Filter for assets which either are, or point to, an IP address owned by the '
            'provided organization.'
        ),
    ] = None,
    whois_email: Annotated[
        Optional[str],
        Doc('Filter for assets where the WHOIS email address matches the provided value.'),
    ] = None,
    whois_email_current: Annotated[
        Optional[str],
        Doc(
            'Filter for assets where the WHOIS email address matches the provided value on '
            'the current WHOIS record.'
        ),
    ] = None,
    open_port_number: Annotated[
        Optional[int],
        Doc('Filter for assets which have an open port with the provided number.'),
    ] = None,
    open_port_protocol: Annotated[
        Optional[str],
        Doc('Filter for assets which have an open port on the provided protocol.'),
    ] = None,
    open_port_service: Annotated[
        Optional[str],
        Doc(
            'Filter for assets which have an open port that appears to support the provided '
            'protocol.'
        ),
    ] = None,
    open_port_technology: Annotated[
        Optional[str],
        Doc('Filter for assets which have a specific product listening on an open port.'),
    ] = None,
    technology_name: Annotated[
        Optional[str],
        Doc('Filter for the name of a technology found on the asset.'),
    ] = None,
    web_technology_name: Annotated[
        Optional[str],
        Doc(
            'Filter for the name of a technology specifically associated with web '
            'resources, such as jQuery or Wordpress.'
        ),
    ] = None,
    certificate_issuer: Annotated[
        Optional[str],
        Doc(
            "Filter where the certificate issuer's common name or organization matches the "
            'provided value.'
        ),
    ] = None,
    certificate_expires_before: Annotated[
        Optional[str],
        Doc('Filter where the certificate expiration date is before the provided value.'),
    ] = None,
    certificate_expires_after: Annotated[
        Optional[str],
        Doc('Filter where the certificate expiration date is after the provided value.'),
    ] = None,
    certificate_issued_before: Annotated[
        Optional[str],
        Doc('Filter where the certificate issuance date is before the provided value.'),
    ] = None,
    certificate_issued_after: Annotated[
        Optional[str],
        Doc('Filter where the certificate issuance date is after the provided value.'),
    ] = None,
    certificate_subject: Annotated[
        Optional[str],
        Doc('Filter where certificate subject or organizationName matches the value.'),
    ] = None,
    certificate_subject_alt_name: Annotated[
        Optional[str],
        Doc('Filter where the certificate Subject Alternative Name matches the value.'),
    ] = None,
    certificate_sha256: Annotated[
        Optional[str],
        Doc('Filter where the certificate public key sha256 value matches the value.'),
    ] = None,
    certificate_covers_domain: Annotated[
        Optional[str],
        Doc(
            'Filter where the certificate subject common name or SAN exactly matches or '
            'wildcard-covers the provided value.'
        ),
    ] = None,
    waf_detected: Annotated[
        Optional[bool],
        Doc('Filter for assets where a WAF is detected.'),
    ] = None,
    waf_name: Annotated[
        Optional[str],
        Doc('Filter for assets where a specific WAF is detected.'),
    ] = None,
    is_responsive: Annotated[
        Optional[bool],
        Doc(
            'Filter for assets that are either responsive or not responsive over ICMP and '
            'port scanning.'
        ),
    ] = None,
    exposure_score_gte: Annotated[
        Optional[int],
        Field(ge=0, le=100),
        Doc('Filter for assets with exposure score greater than or equal to this value.'),
    ] = None,
    exposure_score_lte: Annotated[
        Optional[int],
        Field(ge=0, le=100),
        Doc('Filter for assets with exposure score less than or equal to this value.'),
    ] = None,
    exposure_severity: Annotated[
        Optional[Literal['unknown', 'informational', 'moderate', 'critical']],
        Doc(
            'Filter for assets with an exposure severity matching or higher than the '
            'provided value.'
        ),
    ] = None,
    exposure_id: Annotated[
        Optional[str],
        Doc('Filter for assets which have an exposure with the provided ASI Signature ID.'),
    ] = None,
    additional_fields: Annotated[
        Optional[
            list[
                Literal[
                    'custom_tags',
                    'dns_records',
                    'whois',
                    'ip_metadata',
                    'open_tcp_ports',
                    'open_udp_ports',
                    'web_technologies',
                    'certificates',
                    'certificate_chain',
                    'defenses',
                    'exposures',
                    'exposure_instance_details',
                ]
            ]
        ],
        Doc(
            'Additional fields to include in the response. May be specified multiple times '
            'or as a comma-separated list in the raw API.'
        ),
    ] = None,
    max_results: Annotated[
        Optional[int], Doc('Maximum number of assets to fetch')
    ] = DEFAULT_LIMIT,
    assets_per_page: Annotated[Optional[int], Doc('Results per page')] = DEFAULT_LIMIT,
) -> Annotated[AssetResponse, Doc('Response model for ASI assets list')]:
    """Fetch assets within an ASI project.

    Does pagination requests on batches of the API default page size up to `max_results`.

    Endpoint:
        `v2/projects/{project_id}/assets`

    Raises:
        ValidationError: If response data does not match the `AssetResponse` model.
        ASIFetchAssetError: If an API or connection error occurs.
    """
    params = {
        k: v
        for k, v in locals().items()
        if k not in ('self', 'max_results', 'assets_per_page', 'project_id')
    }
    data = self.asi_client.request_paged(
        'GET',
        EP_ASI_ASSETS.format(project_id),
        params=params,
        max_results=max_results,
        objects_per_page=assets_per_page,
    )

    return AssetResponse.model_validate({'data': data['data'], 'meta': data['meta']})

fetch_asset

fetch_asset(
    project_id: str,
    asset_id: str,
    additional_fields: Optional[
        list[
            Literal[
                'custom_tags',
                'dns_records',
                'whois',
                'ip_metadata',
                'open_tcp_ports',
                'open_udp_ports',
                'web_technologies',
                'certificates',
                'certificate_chain',
                'defenses',
                'exposures',
                'exposure_instance_details',
            ]
        ]
    ] = None,
) -> Asset

Fetch a single asset within an ASI project.

PARAMETER	DESCRIPTION
project_id	The ID of the ASI project to search assets within TYPE: str
asset_id	The asset ID to search for. TYPE: str
additional_fields	Additional fields to include in the response. May be specified multiple times or as a comma-separated list in the raw API. TYPE: Optional[list[Literal['custom_tags', 'dns_records', 'whois', 'ip_metadata', 'open_tcp_ports', 'open_udp_ports', 'web_technologies', 'certificates', 'certificate_chain', 'defenses', 'exposures', 'exposure_instance_details']]] DEFAULT: None

Endpoint

v2/projects/{project_id}/assets/{asset_id}

RAISES	DESCRIPTION
ValidationError	If response data does not match the Asset model.
ASIFetchAssetError	If an API or connection error occurs.

RETURNS	DESCRIPTION
Asset	ASI asset model for the requested asset

Source code in psengine/asi/asi_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ASIFetchAssetError)
def fetch_asset(
    self,
    project_id: Annotated[str, Doc('The ID of the ASI project to search assets within')],
    asset_id: Annotated[str, Doc('The asset ID to search for.')],
    additional_fields: Annotated[
        Optional[
            list[
                Literal[
                    'custom_tags',
                    'dns_records',
                    'whois',
                    'ip_metadata',
                    'open_tcp_ports',
                    'open_udp_ports',
                    'web_technologies',
                    'certificates',
                    'certificate_chain',
                    'defenses',
                    'exposures',
                    'exposure_instance_details',
                ]
            ]
        ],
        Doc(
            'Additional fields to include in the response. May be specified multiple times '
            'or as a comma-separated list in the raw API.'
        ),
    ] = None,
) -> Annotated[Asset, Doc('ASI asset model for the requested asset')]:
    """Fetch a single asset within an ASI project.

    Endpoint:
        `v2/projects/{project_id}/assets/{asset_id}`

    Raises:
        ValidationError: If response data does not match the `Asset` model.
        ASIFetchAssetError: If an API or connection error occurs.
    """
    params = {k: v for k, v in locals().items() if k not in ('self', 'project_id', 'asset_id')}
    data = self.asi_client.request(
        'GET',
        EP_ASI_ASSET.format(project_id, asset_id),
        params=params,
    ).json()

    return Asset.model_validate(data['data'])

fetch_asset_exposures

fetch_asset_exposures(
    project_id: str, asset_id: str
) -> AssetExposuresOut

Fetch exposures for a single asset within an ASI project.

PARAMETER	DESCRIPTION
project_id	The ID of the ASI project to search assets within TYPE: str
asset_id	The asset ID to search for. TYPE: str

Endpoint

v2/projects/{project_id}/assets/{asset_id}/exposures

RAISES	DESCRIPTION
ValidationError	If response data does not match the Asset model.
ASIFetchAssetError	If an API or connection error occurs.

RETURNS	DESCRIPTION
AssetExposuresOut	ASI asset model including exposures for the requested asset

Source code in psengine/asi/asi_mgr.py

@debug_call
@validate_call
@connection_exceptions(ignore_status_code=[], exception_to_raise=ASIFetchAssetError)
def fetch_asset_exposures(
    self,
    project_id: Annotated[str, Doc('The ID of the ASI project to search assets within')],
    asset_id: Annotated[str, Doc('The asset ID to search for.')],
) -> Annotated[
    AssetExposuresOut, Doc('ASI asset model including exposures for the requested asset')
]:
    """Fetch exposures for a single asset within an ASI project.

    Endpoint:
        `v2/projects/{project_id}/assets/{asset_id}/exposures`

    Raises:
        ValidationError: If response data does not match the `Asset` model.
        ASIFetchAssetError: If an API or connection error occurs.
    """
    response = self.asi_client.request(
        'GET',
        EP_ASI_ASSET_EXPOSURES.format(project_id, asset_id),
    ).json()

    return AssetExposuresOut.model_validate(
        {'data': response['data'], 'meta': response['meta']}
    )