create_code_finding ¶

def create_code_finding(
    self,
    environment_id: "capo_continuum.types.environment_id.EnvironmentId",
    ocsf_document: object,
    *,
    config_overrides: Optional[ContinuumClientConfig] = None,
    client_token: Optional[
        "capo_continuum.types.finding_client_token.FindingClientToken"
    ] = None,
) -> "capo_continuum.types.create_code_finding_output.CreateCodeFindingOutput":
    """Ingests a customer-supplied security finding into an environment. Continuum validates the document synchronously against the required overlay fields for the declared format; invalid documents are rejected with a ValidationException. The maximum accepted document size is 256 KB. The environment must have a source configured, and the finding must identify a code location so Continuum can tie it to a repository in the environment's build. Continuum assigns a findingId and stores the verbatim document. When no successful build yet contains the finding's repository, the finding stays SUBMITTED until the next successful build; if that build does not contain it either, the finding is DISCARDED with discardReason SOURCE_REPO_NOT_FOUND.

    Args:
        environment_id: The environment to ingest the finding into.
        ocsf_document: The OCSF finding document to ingest, verbatim. Must declare a class_uid Continuum supports (2002 Vulnerability Finding, or 2007 Application Security Posture Finding) and validate against the vendored OCSF JSON Schema for that class, including the Continuum overlay requirements that finding_info.uid be present and that vulnerabilities[0].affected_code[0].file identify the code location. With several repositories in the environment, the file path must begin with the repository name. Maximum size 256 KB. Stored verbatim and normalized into the typed Finding members returned by GetCodeFinding. GetCodeFinding does not return the document itself.
        client_token: Caller-provided idempotency token. A retry with the same token and an identical finding document replays the original result; the same token with different content is rejected.

    Raises:
        capo_continuum.errors.access_denied_exception.AccessDeniedException: The request is not authorized. Verify that your credentials and request signature are valid and that you have permission to perform this action.
        capo_continuum.errors.conflict_exception.ConflictException: The request could not be completed due to a conflict with the current state of the resource.
        capo_continuum.errors.internal_server_exception.InternalServerException: The request couldn't be processed because of an internal server issue. Try again later. If the problem persists, contact AWS Support.
        capo_continuum.errors.resource_not_found_exception.ResourceNotFoundException: The specified resource was not found. Verify that the resource identifier is correct and that the resource exists.
        capo_continuum.errors.throttling_exception.ThrottlingException: The request was denied due to request throttling.
        capo_continuum.errors.validation_exception.ValidationException
        capo_continuum.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_continuum.types.create_code_finding_input.CreateCodeFindingInput]",
    ) -> OperationResponse[
        "capo_continuum.types.create_code_finding_output.CreateCodeFindingOutput"
    ]:
        import capo_continuum._operations.continuum_control_plane.create_code_finding

        output, http_response = (
            capo_continuum._operations.continuum_control_plane.create_code_finding.create_code_finding(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    interceptors_, options_ = self.operation_options(
        config_overrides, OperationLogging("CreateCodeFinding")
    )
    input_: capo_continuum.types.create_code_finding_input.CreateCodeFindingInput = {
        "environment_id": environment_id,
        "ocsf_document": ocsf_document,
    }
    if client_token is None:
        client_token = str(uuid.uuid4())
    input_["client_token"] = client_token

    response = execute_pipeline(
        OperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    response.response.close()
    return response.output