create_pentest async ¶

async def create_pentest(
    self,
    title: str,
    agent_space_id: str,
    *,
    config_overrides: Optional[AsyncSecurityAgentClientConfig] = None,
    assets: Optional["capo_securityagent.types.assets.Assets"] = None,
    exclude_risk_types: Optional[
        "capo_securityagent.types.risk_type_list.RiskTypeList"
    ] = None,
    service_role: Optional[
        "capo_securityagent.types.service_role.ServiceRole"
    ] = None,
    log_config: Optional[
        "capo_securityagent.types.cloud_watch_log.CloudWatchLog"
    ] = None,
    vpc_config: Optional["capo_securityagent.types.vpc_config.VpcConfig"] = None,
    network_traffic_config: Optional[
        "capo_securityagent.types.network_traffic_config.NetworkTrafficConfig"
    ] = None,
    code_remediation_strategy: Optional[
        "capo_securityagent.types.code_remediation_strategy.CodeRemediationStrategy"
    ] = None,
    disable_managed_skills: Optional[
        "capo_securityagent.types.skill_type_list.SkillTypeList"
    ] = None,
    max_task_hours: Optional[float] = None,
    report_destination: Optional[
        "capo_securityagent.types.report_destination.ReportDestination"
    ] = None,
    report_filters: Optional[
        "capo_securityagent.types.report_filters.ReportFilters"
    ] = None,
    cicd_configuration: Optional[
        "capo_securityagent.types.ci_cd_configuration.CiCdConfiguration"
    ] = None,
    test_scope: Optional["capo_securityagent.types.test_scope.TestScope"] = None,
) -> "capo_securityagent.types.create_pentest_output.CreatePentestOutput":
    """<p>Creates a new pentest configuration in an agent space. A pentest defines the security test parameters, including target assets, risk type exclusions, and logging configuration.</p>

    Args:
        title: <p>The title of the pentest.</p>
        agent_space_id: <p>The unique identifier of the agent space to create the pentest in.</p>
        assets: <p>The assets to include in the pentest, such as endpoints, actors, documents, and source code.</p>
        exclude_risk_types: <p>The list of risk types to exclude from the pentest.</p>
        service_role: <p>The IAM service role to use for the pentest.</p>
        log_config: <p>The CloudWatch Logs configuration for the pentest.</p>
        vpc_config: <p>The VPC configuration for the pentest.</p>
        network_traffic_config: <p>The network traffic configuration for the pentest, including custom headers and traffic rules.</p>
        code_remediation_strategy: <p>The code remediation strategy for the pentest. Valid values are AUTOMATIC and DISABLED.</p>
        disable_managed_skills: <p>A list of managed skills to disable for this pentest. Valid values include FINDING_PERSONALIZATION and LOGIN_OPTIMIZATION.</p>
        max_task_hours: <p>The maximum number of billable task hours allowed for jobs started from this pentest. Must be a positive number. If not set, jobs run to completion with no budget cap.</p>
        report_destination: <p>The destination for publishing scan reports to an integrated document provider.</p>
        report_filters: <p>The report-generation filters applied when the report is exported.</p>
        cicd_configuration: <p>The CI/CD pentesting configuration to apply to the pentest.</p>

    Raises:
        capo_securityagent.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    async def _handler(
        req: "AsyncOperationRequest[capo_securityagent.types.create_pentest_input.CreatePentestInput]",
    ) -> AsyncOperationResponse[
        "capo_securityagent.types.create_pentest_output.CreatePentestOutput"
    ]:
        import capo_securityagent._operations.security_agent.create_pentest

        (
            output,
            http_response,
        ) = await capo_securityagent._operations.security_agent.create_pentest.async_create_pentest(
            req.options, req.input
        )
        return AsyncOperationResponse(output=output, response=http_response)

    def _redact_input(
        input_: "capo_securityagent.types.create_pentest_input.CreatePentestInput",
    ) -> dict[str, Any]:
        import capo_securityagent.types.create_pentest_input

        return capo_securityagent.types.create_pentest_input.redact_sensitive(
            input_
        )

    def _redact_output(
        output: "capo_securityagent.types.create_pentest_output.CreatePentestOutput",
    ) -> dict[str, Any]:
        import capo_securityagent.types.create_pentest_output

        return capo_securityagent.types.create_pentest_output.redact_sensitive(
            output
        )

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging(
            "CreatePentest",
            redact_input=_redact_input,
            redact_output=_redact_output,
        ),
    )
    input_: capo_securityagent.types.create_pentest_input.CreatePentestInput = {
        "title": title,
        "agent_space_id": agent_space_id,
    }
    if assets is not None:
        input_["assets"] = assets
    if exclude_risk_types is not None:
        input_["exclude_risk_types"] = exclude_risk_types
    if service_role is not None:
        input_["service_role"] = service_role
    if log_config is not None:
        input_["log_config"] = log_config
    if vpc_config is not None:
        input_["vpc_config"] = vpc_config
    if network_traffic_config is not None:
        input_["network_traffic_config"] = network_traffic_config
    if code_remediation_strategy is not None:
        input_["code_remediation_strategy"] = code_remediation_strategy
    if disable_managed_skills is not None:
        input_["disable_managed_skills"] = disable_managed_skills
    if max_task_hours is not None:
        input_["max_task_hours"] = max_task_hours
    if report_destination is not None:
        input_["report_destination"] = report_destination
    if report_filters is not None:
        input_["report_filters"] = report_filters
    if cicd_configuration is not None:
        input_["cicd_configuration"] = cicd_configuration
    if test_scope is not None:
        input_["test_scope"] = test_scope

    response = await aexecute_pipeline(
        AsyncOperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    await response.response.aclose()
    return response.output