API Python Reference

This page provides auto-generated documentation for the SecretZero API module, including all FastAPI endpoints, schemas, and utilities.

Overview

The SecretZero API is built with FastAPI and provides a REST interface for all SecretZero operations. This reference documents the Python implementation details.

Interactive API Documentation

For interactive API testing and OpenAPI specifications, see the Interactive API Reference.

API Application

app

FastAPI application for SecretZero API.

create_app(secretfile_path='Secretfile.yml')

Create and configure the FastAPI application.

Parameters:

Name	Type	Description	Default
secretfile_path	str	Path to the Secretfile configuration	'Secretfile.yml'

Returns:

Type	Description
FastAPI	Configured FastAPI application

Source code in src/secretzero/api/app.py

def create_app(secretfile_path: str = "Secretfile.yml") -> FastAPI:
    """Create and configure the FastAPI application.

    Args:
        secretfile_path: Path to the Secretfile configuration

    Returns:
        Configured FastAPI application
    """
    app = FastAPI(
        title="SecretZero API",
        description="REST API for SecretZero secrets orchestration and lifecycle management",
        version=__version__,
        docs_url="/docs",
        redoc_url="/redoc",
        openapi_url="/openapi.json",
    )

    # CORS middleware - DEVELOPMENT ONLY
    # TODO: Configure restrictive CORS policy for production
    # For production, set allow_origins to specific domains:
    # allow_origins=["https://yourdomain.com"]
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],  # ⚠️ SECURITY WARNING: Allow all origins (development only)
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Store configuration path
    app.state.secretfile_path = secretfile_path

    @app.exception_handler(Exception)
    async def global_exception_handler(request, exc):
        """Handle all uncaught exceptions."""
        audit_logger = get_audit_logger()
        audit_logger.log(
            action="error",
            resource=str(request.url),
            details={"error": str(exc)},
            success=False,
        )
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content=ErrorResponse(
                error="Internal server error",
                detail=str(exc),
            ).model_dump(),
        )

    @app.get("/", response_model=dict[str, str])
    async def root():
        """Root endpoint."""
        return {
            "message": "SecretZero API",
            "version": __version__,
            "docs": "/docs",
        }

    @app.get("/health", response_model=HealthResponse)
    async def health():
        """Health check endpoint."""
        return HealthResponse(version=__version__)

    @app.post("/config/validate", response_model=ConfigValidationResponse)
    async def validate_config(
        request: ConfigValidationRequest,
        _auth: str = RequireAuth,
    ):
        """Validate a Secretfile configuration."""
        audit_logger = get_audit_logger()

        try:
            # Try to load and validate the config
            # For now, we just check if it can be parsed
            errors = []
            warnings = []

            # Basic validation
            if "version" not in request.config:
                errors.append("Missing required field: version")
            if "secrets" not in request.config:
                errors.append("Missing required field: secrets")

            valid = len(errors) == 0

            audit_logger.log(
                action="validate_config",
                resource="config",
                details={"valid": valid, "errors": errors},
                success=valid,
            )

            return ConfigValidationResponse(
                valid=valid,
                errors=errors,
                warnings=warnings,
            )
        except Exception as e:
            audit_logger.log(
                action="validate_config",
                resource="config",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail=f"Configuration validation failed: {str(e)}",
            )

    @app.get("/secrets", response_model=SecretListResponse)
    async def list_secrets(_auth: str = RequireAuth):
        """List all secrets from the Secretfile."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            loader = ConfigLoader()
            config = loader.load_file(config_path)

            secrets = []
            for secret in config.secrets:
                secrets.append(
                    {
                        "name": secret.name,
                        "kind": secret.kind,
                        "rotation_period": secret.rotation_period,
                        "targets": [t.kind for t in secret.targets],
                    }
                )

            audit_logger.log(
                action="list_secrets",
                resource="secrets",
                details={"count": len(secrets)},
            )

            return SecretListResponse(secrets=secrets, count=len(secrets))
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="list_secrets",
                resource="secrets",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Failed to list secrets: {str(e)}",
            )

    @app.get("/secrets/{secret_name}/status", response_model=SecretStatusResponse)
    async def get_secret_status(secret_name: str, _auth: str = RequireAuth):
        """Get the status of a specific secret."""
        audit_logger = get_audit_logger()

        try:
            lockfile = Lockfile.load(Path(".gitsecrets.lock"))
            entry = lockfile.get_secret_info(secret_name)

            if entry is None:
                audit_logger.log(
                    action="get_secret_status",
                    resource=f"secret:{secret_name}",
                    details={"exists": False},
                )
                return SecretStatusResponse(name=secret_name, exists=False)

            audit_logger.log(
                action="get_secret_status",
                resource=f"secret:{secret_name}",
                details={"exists": True},
            )

            return SecretStatusResponse(
                name=secret_name,
                exists=True,
                created_at=entry.created_at,
                updated_at=entry.updated_at,
                last_rotated=entry.last_rotated,
                rotation_count=entry.rotation_count,
                targets=list(entry.targets.keys()),
            )
        except Exception as e:
            audit_logger.log(
                action="get_secret_status",
                resource=f"secret:{secret_name}",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Failed to get secret status: {str(e)}",
            )

    @app.post("/sync", response_model=SyncResponse)
    async def sync_secrets(request: SyncRequest, _auth: str = RequireAuth):
        """Sync secrets (generate and store)."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            loader = ConfigLoader()
            config = loader.load_file(config_path)
            lockfile = Lockfile.load(Path(".gitsecrets.lock"))
            sync_engine = SyncEngine(config, lockfile)

            generated = []
            skipped = []

            if request.secret_name:
                # Sync specific secret
                secret = next((s for s in config.secrets if s.name == request.secret_name), None)
                if not secret:
                    raise HTTPException(
                        status_code=status.HTTP_404_NOT_FOUND,
                        detail=f"Secret not found: {request.secret_name}",
                    )

                result = sync_engine._sync_secret(secret, request.dry_run, request.force)
                if not request.dry_run and result["generated"] and result["stored"]:
                    generated.append(request.secret_name)
                elif request.dry_run or result["skipped"]:
                    skipped.append(request.secret_name)
                message = f"{'Would generate' if request.dry_run else 'Generated'} secret: {request.secret_name}"
            else:
                # Sync all secrets
                results = sync_engine.sync(dry_run=request.dry_run, force_rotation=request.force)
                for detail in results.get("details", []):
                    if not request.dry_run and detail.get("generated") and detail.get("stored"):
                        generated.append(detail["name"])
                    elif request.dry_run or detail.get("skipped"):
                        skipped.append(detail["name"])
                message = f"{'Would generate' if request.dry_run else 'Generated'} {len(generated)} secret(s), skipped {len(skipped)}"

            # Save lockfile if not dry run
            if not request.dry_run:
                lockfile.save(Path(".gitsecrets.lock"))

            audit_logger.log(
                action="sync",
                resource="secrets",
                details={
                    "dry_run": request.dry_run,
                    "force": request.force,
                    "generated": generated,
                    "skipped": skipped,
                },
            )

            return SyncResponse(
                secrets_generated=generated,
                secrets_skipped=skipped,
                message=message,
            )
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="sync",
                resource="secrets",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Sync failed: {str(e)}",
            )

    @app.post("/rotation/check", response_model=RotationCheckResponse)
    async def check_rotation(request: RotationCheckRequest, _auth: str = RequireAuth):
        """Check which secrets need rotation."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            loader = ConfigLoader()
            config = loader.load_file(config_path)
            lockfile = Lockfile.load(Path(".gitsecrets.lock"))

            secrets_to_check = []
            if request.secret_name:
                secret = next((s for s in config.secrets if s.name == request.secret_name), None)
                if not secret:
                    raise HTTPException(
                        status_code=status.HTTP_404_NOT_FOUND,
                        detail=f"Secret not found: {request.secret_name}",
                    )
                secrets_to_check = [secret]
            else:
                secrets_to_check = config.secrets

            due = []
            overdue = []
            results = []

            for secret in secrets_to_check:
                entry = lockfile.get_secret_info(secret.name)
                if entry and secret.rotation_period:
                    should_rotate, status_msg = should_rotate_secret(secret, entry)
                    results.append(
                        {
                            "name": secret.name,
                            "should_rotate": should_rotate,
                            "status": status_msg,
                        }
                    )
                    if should_rotate:
                        if "overdue" in status_msg.lower():
                            overdue.append(secret.name)
                        else:
                            due.append(secret.name)

            audit_logger.log(
                action="check_rotation",
                resource="secrets",
                details={
                    "secrets_checked": len(secrets_to_check),
                    "secrets_due": len(due),
                    "secrets_overdue": len(overdue),
                },
            )

            return RotationCheckResponse(
                secrets_checked=len(secrets_to_check),
                secrets_due=due,
                secrets_overdue=overdue,
                results=results,
            )
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="check_rotation",
                resource="secrets",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Rotation check failed: {str(e)}",
            )

    @app.post("/rotation/execute", response_model=RotationExecuteResponse)
    async def execute_rotation(request: RotationExecuteRequest, _auth: str = RequireAuth):
        """Execute secret rotation."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            loader = ConfigLoader()
            config = loader.load_file(config_path)
            lockfile = Lockfile.load(Path(".gitsecrets.lock"))
            sync_engine = SyncEngine(config, lockfile)

            rotated = []
            failed = []

            secrets_to_rotate = []
            if request.secret_name:
                secret = next((s for s in config.secrets if s.name == request.secret_name), None)
                if not secret:
                    raise HTTPException(
                        status_code=status.HTTP_404_NOT_FOUND,
                        detail=f"Secret not found: {request.secret_name}",
                    )
                secrets_to_rotate = [secret]
            else:
                # Find all secrets that need rotation
                for secret in config.secrets:
                    entry = lockfile.get_secret_info(secret.name)
                    if entry and secret.rotation_period:
                        should_rotate, _ = should_rotate_secret(secret, entry)
                        if should_rotate or request.force:
                            secrets_to_rotate.append(secret)

            for secret in secrets_to_rotate:
                try:
                    result = sync_engine._sync_secret(secret, dry_run=False, force_rotation=True)
                    if result["generated"]:
                        rotated.append(secret.name)
                    else:
                        failed.append(secret.name)
                except Exception:
                    failed.append(secret.name)

            # Save lockfile
            lockfile.save(Path(".gitsecrets.lock"))

            message = f"Rotated {len(rotated)} secret(s)"
            if failed:
                message += f", {len(failed)} failed"

            audit_logger.log(
                action="execute_rotation",
                resource="secrets",
                details={
                    "force": request.force,
                    "rotated": rotated,
                    "failed": failed,
                },
            )

            return RotationExecuteResponse(
                rotated=rotated,
                failed=failed,
                message=message,
            )
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="execute_rotation",
                resource="secrets",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Rotation execution failed: {str(e)}",
            )

    @app.post("/policy/check", response_model=PolicyCheckResponse)
    async def check_policy(request: PolicyCheckRequest, _auth: str = RequireAuth):
        """Check policy compliance."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            loader = ConfigLoader()
            config = loader.load_file(config_path)

            # Create policy engine
            engine = PolicyEngine(config)
            violations = engine.validate_all()

            errors = []
            warnings = []
            info = []

            for violation in violations:
                entry = {
                    "secret": violation.secret_name,
                    "policy": violation.policy_name,
                    "message": violation.message,
                    "severity": violation.severity,
                }

                if violation.severity == "error":
                    errors.append(entry)
                elif violation.severity == "warning":
                    warnings.append(entry)
                else:
                    info.append(entry)

            compliant = len(errors) == 0 and (not request.fail_on_warning or len(warnings) == 0)

            audit_logger.log(
                action="check_policy",
                resource="secrets",
                details={
                    "compliant": compliant,
                    "errors": len(errors),
                    "warnings": len(warnings),
                },
            )

            return PolicyCheckResponse(
                compliant=compliant,
                errors=errors,
                warnings=warnings,
                info=info,
            )
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="check_policy",
                resource="secrets",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Policy check failed: {str(e)}",
            )

    @app.post("/drift/check", response_model=DriftCheckResponse)
    async def check_drift(request: DriftCheckRequest, _auth: str = RequireAuth):
        """Check for configuration drift."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            # loader = ConfigLoader()
            # config = loader.load_file(config_path)
            lockfile_path = Path(".gitsecrets.lock")
            detector = DriftDetector(config_path, lockfile_path)

            # Check drift
            drift_results = detector.check_drift(request.secret_name)

            secrets_with_drift = []
            details = []

            for drift_status in drift_results:
                if drift_status.has_drift:
                    secrets_with_drift.append(drift_status.secret_name)
                    details.append(
                        {
                            "secret": drift_status.secret_name,
                            "message": drift_status.message,
                            "details": drift_status.details,
                        }
                    )

            has_drift = len(secrets_with_drift) > 0

            audit_logger.log(
                action="check_drift",
                resource="secrets",
                details={
                    "has_drift": has_drift,
                    "secrets_with_drift": secrets_with_drift,
                },
            )

            return DriftCheckResponse(
                has_drift=has_drift,
                secrets_with_drift=secrets_with_drift,
                details=details,
            )
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="check_drift",
                resource="secrets",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Drift check failed: {str(e)}",
            )

    @app.get("/audit/logs", response_model=AuditLogResponse)
    async def get_audit_logs(
        limit: int = 50,
        offset: int = 0,
        action: str | None = None,
        resource: str | None = None,
        _auth: str = RequireAuth,
    ):
        """Get audit logs."""
        audit_logger = get_audit_logger()

        try:
            logs = audit_logger.get_logs(
                limit=limit,
                offset=offset,
                action=action,
                resource=resource,
            )

            return AuditLogResponse(
                entries=logs,
                count=len(logs),
                page=offset // limit + 1,
                per_page=limit,
            )
        except Exception as e:
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Failed to retrieve audit logs: {str(e)}",
            )

    @app.get("/list/providers", response_model=ProviderListResponse)
    async def list_providers(_auth: str = RequireAuth):
        """List all providers configured in the Secretfile."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            loader = ConfigLoader()
            config = loader.load_file(config_path)

            providers = [
                {
                    "name": name,
                    "kind": p.kind,
                    "auth_kind": p.auth.kind if p.auth else None,
                    "fallback_generator": p.fallback_generator,
                }
                for name, p in config.providers.items()
            ]

            audit_logger.log(
                action="list_providers",
                resource="providers",
                details={"count": len(providers)},
            )

            return ProviderListResponse(providers=providers, total=len(providers))
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="list_providers",
                resource="providers",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Failed to list providers: {str(e)}",
            )

    @app.get("/list/targets", response_model=TargetListResponse)
    async def list_targets(_auth: str = RequireAuth):
        """List all target destinations across all secrets in the Secretfile."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            loader = ConfigLoader()
            config = loader.load_file(config_path)

            all_targets = []
            for secret in config.secrets:
                for t in secret.targets:
                    all_targets.append(
                        {
                            "secret": secret.name,
                            "provider": t.provider,
                            "kind": t.kind,
                            "config": t.config,
                        }
                    )

            audit_logger.log(
                action="list_targets",
                resource="targets",
                details={"count": len(all_targets)},
            )

            return TargetListResponse(targets=all_targets, total=len(all_targets))
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="list_targets",
                resource="targets",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Failed to list targets: {str(e)}",
            )

    @app.get("/list/variables", response_model=VariableListResponse)
    async def list_variables(_auth: str = RequireAuth):
        """List all variables defined in the Secretfile."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            loader = ConfigLoader()
            config = loader.load_file(config_path)

            variables = dict(config.variables)

            audit_logger.log(
                action="list_variables",
                resource="variables",
                details={"count": len(variables)},
            )

            return VariableListResponse(variables=variables, total=len(variables))
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="list_variables",
                resource="variables",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Failed to list variables: {str(e)}",
            )

    @app.get("/config/render", response_model=ConfigRenderResponse)
    async def render_config(_auth: str = RequireAuth):
        """Render the Secretfile configuration with variables interpolated."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            loader = ConfigLoader()
            config = loader.load_file(config_path)

            config_dict = config.model_dump(mode="python", exclude_none=True)

            audit_logger.log(
                action="render_config",
                resource="config",
                details={},
            )

            return ConfigRenderResponse(config=config_dict)
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="render_config",
                resource="config",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Failed to render config: {str(e)}",
            )

    @app.get("/schema")
    async def get_schema():
        """Export JSON Schema for Secretfile.yml."""
        return Secretfile.model_json_schema()

    @app.get("/secret-types", response_model=SecretTypesResponse)
    async def get_secret_types():
        """List all available secret generator and target types."""
        generator_types = []
        for name in dir(generators):
            if name.endswith("Generator") and not name.startswith("_"):
                obj = getattr(generators, name)
                if inspect.isclass(obj) and obj != generators.BaseGenerator:
                    type_name = _class_name_to_snake_case(name, "Generator")
                    description = (obj.__doc__ or "").strip().split("\n")[0]
                    generator_types.append({"type": type_name, "description": description})

        target_types = []
        for name in dir(targets):
            if name.endswith("Target") and not name.startswith("_"):
                obj = getattr(targets, name)
                if inspect.isclass(obj) and obj != targets.BaseTarget:
                    type_name = _class_name_to_snake_case(name, "Target")
                    description = (obj.__doc__ or "").strip().split("\n")[0]
                    target_types.append({"type": type_name, "description": description})

        return SecretTypesResponse(
            generators=sorted(generator_types, key=lambda x: x["type"]),
            targets=sorted(target_types, key=lambda x: x["type"]),
        )

    @app.get("/graph", response_model=GraphResponse)
    async def get_graph(_auth: str = RequireAuth):
        """Generate a graph representation of Secretfile relationships."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            loader = ConfigLoader()
            config = loader.load_file(config_path)

            nodes = []
            edges = []
            for secret in config.secrets:
                nodes.append(
                    {
                        "id": secret.name,
                        "type": "secret",
                        "kind": secret.kind,
                        "one_time": secret.one_time,
                        "rotation_period": secret.rotation_period,
                    }
                )
                edges.append(
                    {
                        "from": secret.kind,
                        "to": secret.name,
                        "label": "generates",
                    }
                )
                for target in secret.targets:
                    target_id = f"{target.provider}/{target.kind}"
                    if not any(n["id"] == target_id for n in nodes):
                        nodes.append(
                            {
                                "id": target_id,
                                "type": "target",
                                "provider": target.provider,
                                "kind": target.kind,
                            }
                        )
                    edges.append({"from": secret.name, "to": target_id, "label": "stored_in"})

            audit_logger.log(
                action="get_graph",
                resource="config",
                details={"nodes": len(nodes), "edges": len(edges)},
            )

            return GraphResponse(nodes=nodes, edges=edges)
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="get_graph",
                resource="config",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Failed to generate graph: {str(e)}",
            )

    @app.get("/secrets/{secret_name}", response_model=SecretDetailResponse)
    async def get_secret_detail(secret_name: str, _auth: str = RequireAuth):
        """Get detailed information about a specific secret."""
        audit_logger = get_audit_logger()

        try:
            config_path = Path(app.state.secretfile_path)
            if not config_path.exists():
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secretfile not found: {config_path}",
                )

            loader = ConfigLoader()
            config = loader.load_file(config_path)

            secret = next((s for s in config.secrets if s.name == secret_name), None)
            if not secret:
                raise HTTPException(
                    status_code=status.HTTP_404_NOT_FOUND,
                    detail=f"Secret not found: {secret_name}",
                )

            targets = [
                {"provider": t.provider, "kind": t.kind, "config": t.config} for t in secret.targets
            ]

            lockfile = Lockfile.load(Path(".gitsecrets.lock"))
            entry = lockfile.get_secret_info(secret_name)

            audit_logger.log(
                action="get_secret_detail",
                resource=f"secret:{secret_name}",
                details={"exists": entry is not None},
            )

            return SecretDetailResponse(
                name=secret.name,
                kind=secret.kind,
                one_time=secret.one_time,
                rotation_period=secret.rotation_period,
                targets=targets,
                exists=entry is not None,
                created_at=entry.created_at if entry else None,
                updated_at=entry.updated_at if entry else None,
                last_rotated=entry.last_rotated if entry else None,
                rotation_count=entry.rotation_count if entry else 0,
            )
        except HTTPException:
            raise
        except Exception as e:
            audit_logger.log(
                action="get_secret_detail",
                resource=f"secret:{secret_name}",
                details={"error": str(e)},
                success=False,
            )
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Failed to get secret detail: {str(e)}",
            )

    return app

Request/Response Schemas

Pydantic Models

All API requests and responses use Pydantic models for validation and serialization.

schemas

API request and response models.

HealthResponse

Bases: BaseModel

Health check response.

Source code in src/secretzero/api/schemas.py

class HealthResponse(BaseModel):
    """Health check response."""

    status: str = "healthy"
    version: str
    timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC))

ErrorResponse

Bases: BaseModel

Error response.

Source code in src/secretzero/api/schemas.py

class ErrorResponse(BaseModel):
    """Error response."""

    error: str
    detail: str | None = None
    timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC))

ConfigValidationRequest

Bases: BaseModel

Request to validate a Secretfile configuration.

Source code in src/secretzero/api/schemas.py

class ConfigValidationRequest(BaseModel):
    """Request to validate a Secretfile configuration."""

    config: dict[str, Any]

ConfigValidationResponse

Bases: BaseModel

Response from configuration validation.

Source code in src/secretzero/api/schemas.py

class ConfigValidationResponse(BaseModel):
    """Response from configuration validation."""

    valid: bool
    errors: list[str] = Field(default_factory=list)
    warnings: list[str] = Field(default_factory=list)

ConfigRenderResponse

Bases: BaseModel

Response with rendered Secretfile configuration.

Source code in src/secretzero/api/schemas.py

class ConfigRenderResponse(BaseModel):
    """Response with rendered Secretfile configuration."""

    config: dict[str, Any]

SecretListResponse

Bases: BaseModel

Response listing secrets.

Source code in src/secretzero/api/schemas.py

class SecretListResponse(BaseModel):
    """Response listing secrets."""

    secrets: list[dict[str, Any]]
    count: int

SecretDetailResponse

Bases: BaseModel

Detailed secret information (show command equivalent).

Source code in src/secretzero/api/schemas.py

class SecretDetailResponse(BaseModel):
    """Detailed secret information (show command equivalent)."""

    name: str
    kind: str
    one_time: bool = False
    rotation_period: str | None = None
    targets: list[dict[str, Any]] = Field(default_factory=list)
    exists: bool = False
    created_at: datetime | None = None
    updated_at: datetime | None = None
    last_rotated: datetime | None = None
    rotation_count: int = 0

SecretStatusResponse

Bases: BaseModel

Response with secret status.

Source code in src/secretzero/api/schemas.py

class SecretStatusResponse(BaseModel):
    """Response with secret status."""

    name: str
    exists: bool
    created_at: datetime | None = None
    updated_at: datetime | None = None
    last_rotated: datetime | None = None
    rotation_count: int = 0
    targets: list[str] = Field(default_factory=list)

SyncRequest

Bases: BaseModel

Request to sync secrets.

Source code in src/secretzero/api/schemas.py

class SyncRequest(BaseModel):
    """Request to sync secrets."""

    dry_run: bool = False
    force: bool = False
    secret_name: str | None = None

SyncResponse

Bases: BaseModel

Response from sync operation.

Source code in src/secretzero/api/schemas.py

class SyncResponse(BaseModel):
    """Response from sync operation."""

    secrets_generated: list[str] = Field(default_factory=list)
    secrets_skipped: list[str] = Field(default_factory=list)
    message: str

RotationCheckRequest

Bases: BaseModel

Request to check rotation status.

Source code in src/secretzero/api/schemas.py

class RotationCheckRequest(BaseModel):
    """Request to check rotation status."""

    secret_name: str | None = None
    dry_run: bool = True

RotationCheckResponse

Bases: BaseModel

Response from rotation check.

Source code in src/secretzero/api/schemas.py

class RotationCheckResponse(BaseModel):
    """Response from rotation check."""

    secrets_checked: int
    secrets_due: list[str] = Field(default_factory=list)
    secrets_overdue: list[str] = Field(default_factory=list)
    results: list[dict[str, Any]] = Field(default_factory=list)

RotationExecuteRequest

Bases: BaseModel

Request to execute rotation.

Source code in src/secretzero/api/schemas.py

class RotationExecuteRequest(BaseModel):
    """Request to execute rotation."""

    secret_name: str | None = None
    force: bool = False

RotationExecuteResponse

Bases: BaseModel

Response from rotation execution.

Source code in src/secretzero/api/schemas.py

class RotationExecuteResponse(BaseModel):
    """Response from rotation execution."""

    rotated: list[str] = Field(default_factory=list)
    failed: list[str] = Field(default_factory=list)
    message: str

DriftCheckRequest

Bases: BaseModel

Request to check for drift.

Source code in src/secretzero/api/schemas.py

class DriftCheckRequest(BaseModel):
    """Request to check for drift."""

    secret_name: str | None = None

DriftCheckResponse

Bases: BaseModel

Response from drift check.

Source code in src/secretzero/api/schemas.py

class DriftCheckResponse(BaseModel):
    """Response from drift check."""

    has_drift: bool
    secrets_with_drift: list[str] = Field(default_factory=list)
    details: list[dict[str, Any]] = Field(default_factory=list)

PolicyCheckRequest

Bases: BaseModel

Request to check policy compliance.

Source code in src/secretzero/api/schemas.py

class PolicyCheckRequest(BaseModel):
    """Request to check policy compliance."""

    fail_on_warning: bool = False

PolicyCheckResponse

Bases: BaseModel

Response from policy check.

Source code in src/secretzero/api/schemas.py

class PolicyCheckResponse(BaseModel):
    """Response from policy check."""

    compliant: bool
    errors: list[dict[str, Any]] = Field(default_factory=list)
    warnings: list[dict[str, Any]] = Field(default_factory=list)
    info: list[dict[str, Any]] = Field(default_factory=list)

ProviderListResponse

Bases: BaseModel

Response listing providers from Secretfile.

Source code in src/secretzero/api/schemas.py

class ProviderListResponse(BaseModel):
    """Response listing providers from Secretfile."""

    providers: list[dict[str, Any]]
    total: int

TargetListResponse

Bases: BaseModel

Response listing targets from Secretfile.

Source code in src/secretzero/api/schemas.py

class TargetListResponse(BaseModel):
    """Response listing targets from Secretfile."""

    targets: list[dict[str, Any]]
    total: int

SecretTypesResponse

Bases: BaseModel

Response listing available secret generator and target types.

Source code in src/secretzero/api/schemas.py

class SecretTypesResponse(BaseModel):
    """Response listing available secret generator and target types."""

    generators: list[dict[str, str]]
    targets: list[dict[str, str]]

VariableListResponse

Bases: BaseModel

Response listing variables from Secretfile.

Source code in src/secretzero/api/schemas.py

class VariableListResponse(BaseModel):
    """Response listing variables from Secretfile."""

    variables: dict[str, Any]
    total: int

GraphResponse

Bases: BaseModel

Response with graph representation of Secretfile relationships.

Source code in src/secretzero/api/schemas.py

class GraphResponse(BaseModel):
    """Response with graph representation of Secretfile relationships."""

    nodes: list[dict[str, Any]]
    edges: list[dict[str, Any]]

AuditLogResponse

Bases: BaseModel

Response with audit logs.

Source code in src/secretzero/api/schemas.py

class AuditLogResponse(BaseModel):
    """Response with audit logs."""

    entries: list[AuditLogEntry]
    count: int
    page: int = 1
    per_page: int = 50

Authentication

API key authentication middleware.

auth

Authentication and security for the API.

get_api_key_from_env()

Get the API key from environment variable.

Source code in src/secretzero/api/auth.py

def get_api_key_from_env() -> str | None:
    """Get the API key from environment variable."""
    return os.environ.get("SECRETZERO_API_KEY")

generate_api_key()

Generate a random API key.

Source code in src/secretzero/api/auth.py

def generate_api_key() -> str:
    """Generate a random API key."""
    return secrets.token_urlsafe(32)

hash_api_key(api_key)

Hash an API key for comparison.

Source code in src/secretzero/api/auth.py

def hash_api_key(api_key: str) -> str:
    """Hash an API key for comparison."""
    return hashlib.sha256(api_key.encode()).hexdigest()

verify_api_key(api_key=Security(api_key_header)) async

Verify the API key from the request header.

Parameters:

Name	Type	Description	Default
api_key	str	The API key from the request header.	Security(api_key_header)

Returns:

Type	Description
str	The verified API key.

Raises:

Type	Description
HTTPException	If the API key is invalid or missing.

Source code in src/secretzero/api/auth.py

async def verify_api_key(api_key: str = Security(api_key_header)) -> str:
    """Verify the API key from the request header.

    Args:
        api_key: The API key from the request header.

    Returns:
        The verified API key.

    Raises:
        HTTPException: If the API key is invalid or missing.
    """
    # If no API key is configured, allow all requests (insecure mode for testing)
    env_api_key = get_api_key_from_env()
    if env_api_key is None:
        # Running in insecure mode
        return "insecure-mode"

    if api_key is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing API key",
            headers={"WWW-Authenticate": "ApiKey"},
        )

    # Compare hashes for timing attack resistance
    provided_hash = hash_api_key(api_key)
    expected_hash = hash_api_key(env_api_key)

    if not secrets.compare_digest(provided_hash, expected_hash):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid API key",
            headers={"WWW-Authenticate": "ApiKey"},
        )

    return api_key

Audit Logging

Audit trail functionality for API operations.

audit

Audit logging for API operations.

AuditLogger

Simple file-based audit logger.

Source code in src/secretzero/api/audit.py

class AuditLogger:
    """Simple file-based audit logger."""

    def __init__(self, log_file: Path | None = None):
        """Initialize the audit logger.

        Args:
            log_file: Path to the audit log file. Defaults to .secretzero_audit.log
        """
        if log_file is None:
            log_file = Path.cwd() / ".secretzero_audit.log"
        self.log_file = log_file

    def log(
        self,
        action: str,
        resource: str,
        user: str | None = None,
        details: dict[str, Any] | None = None,
        success: bool = True,
    ) -> None:
        """Log an audit event.

        Args:
            action: The action performed (e.g., "sync", "rotate", "policy_check")
            resource: The resource affected (e.g., "secret:api_key")
            user: Optional user identifier
            details: Optional additional details
            success: Whether the operation was successful
        """
        entry = AuditLogEntry(
            action=action,
            resource=resource,
            user=user or "api",
            details=details,
            success=success,
        )

        # Append to log file
        with open(self.log_file, "a") as f:
            f.write(json.dumps(entry.model_dump(mode="json")) + "\n")

    def get_logs(
        self,
        limit: int = 100,
        offset: int = 0,
        action: str | None = None,
        resource: str | None = None,
    ) -> list[AuditLogEntry]:
        """Retrieve audit logs.

        Args:
            limit: Maximum number of logs to return
            offset: Number of logs to skip
            action: Filter by action
            resource: Filter by resource

        Returns:
            List of audit log entries
        """
        if not self.log_file.exists():
            return []

        logs = []
        with open(self.log_file) as f:
            for line in f:
                try:
                    data = json.loads(line.strip())
                    entry = AuditLogEntry(**data)

                    # Apply filters
                    if action and entry.action != action:
                        continue
                    if resource and entry.resource != resource:
                        continue

                    logs.append(entry)
                except (json.JSONDecodeError, ValueError):
                    # Skip invalid entries
                    continue

        # Reverse to get newest first
        logs.reverse()

        # Apply pagination
        return logs[offset : offset + limit]

__init__(log_file=None)

Initialize the audit logger.

Parameters:

Name	Type	Description	Default
log_file	Path | None	Path to the audit log file. Defaults to .secretzero_audit.log	None

Source code in src/secretzero/api/audit.py

def __init__(self, log_file: Path | None = None):
    """Initialize the audit logger.

    Args:
        log_file: Path to the audit log file. Defaults to .secretzero_audit.log
    """
    if log_file is None:
        log_file = Path.cwd() / ".secretzero_audit.log"
    self.log_file = log_file

log(action, resource, user=None, details=None, success=True)

Log an audit event.

Parameters:

Name	Type	Description	Default
action	str	The action performed (e.g., "sync", "rotate", "policy_check")	required
resource	str	The resource affected (e.g., "secret:api_key")	required
user	str | None	Optional user identifier	None
details	dict[str, Any] | None	Optional additional details	None
success	bool	Whether the operation was successful	True

Source code in src/secretzero/api/audit.py

def log(
    self,
    action: str,
    resource: str,
    user: str | None = None,
    details: dict[str, Any] | None = None,
    success: bool = True,
) -> None:
    """Log an audit event.

    Args:
        action: The action performed (e.g., "sync", "rotate", "policy_check")
        resource: The resource affected (e.g., "secret:api_key")
        user: Optional user identifier
        details: Optional additional details
        success: Whether the operation was successful
    """
    entry = AuditLogEntry(
        action=action,
        resource=resource,
        user=user or "api",
        details=details,
        success=success,
    )

    # Append to log file
    with open(self.log_file, "a") as f:
        f.write(json.dumps(entry.model_dump(mode="json")) + "\n")

get_logs(limit=100, offset=0, action=None, resource=None)

Retrieve audit logs.

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of logs to return	100
offset	int	Number of logs to skip	0
action	str | None	Filter by action	None
resource	str | None	Filter by resource	None

Returns:

Type	Description
list[AuditLogEntry]	List of audit log entries

Source code in src/secretzero/api/audit.py

def get_logs(
    self,
    limit: int = 100,
    offset: int = 0,
    action: str | None = None,
    resource: str | None = None,
) -> list[AuditLogEntry]:
    """Retrieve audit logs.

    Args:
        limit: Maximum number of logs to return
        offset: Number of logs to skip
        action: Filter by action
        resource: Filter by resource

    Returns:
        List of audit log entries
    """
    if not self.log_file.exists():
        return []

    logs = []
    with open(self.log_file) as f:
        for line in f:
            try:
                data = json.loads(line.strip())
                entry = AuditLogEntry(**data)

                # Apply filters
                if action and entry.action != action:
                    continue
                if resource and entry.resource != resource:
                    continue

                logs.append(entry)
            except (json.JSONDecodeError, ValueError):
                # Skip invalid entries
                continue

    # Reverse to get newest first
    logs.reverse()

    # Apply pagination
    return logs[offset : offset + limit]

get_audit_logger()

Get the global audit logger instance.

Source code in src/secretzero/api/audit.py

def get_audit_logger() -> AuditLogger:
    """Get the global audit logger instance."""
    global _audit_logger
    if _audit_logger is None:
        _audit_logger = AuditLogger()
    return _audit_logger

Server

CLI entry point for running the API server.

server

Server entry point for the SecretZero API.

run()

Run the SecretZero API server.

Source code in src/secretzero/api/server.py

def run() -> None:
    """Run the SecretZero API server."""
    # Get configuration from environment
    host = os.environ.get("SECRETZERO_HOST", "127.0.0.1")
    port = int(os.environ.get("SECRETZERO_PORT", "8000"))
    secretfile_path = os.environ.get("SECRETZERO_CONFIG", "Secretfile.yml")
    reload = os.environ.get("SECRETZERO_RELOAD", "false").lower() == "true"

    # Check if Secretfile exists
    config_path = Path(secretfile_path)
    if not config_path.exists():
        print(f"Warning: Secretfile not found at {secretfile_path}")
        print("Some API endpoints may not work correctly.")

    # Create app
    app = create_app(secretfile_path=secretfile_path)

    # Print startup info
    print("\n🔐 SecretZero API Server")
    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━")
    print(f"Host:        {host}")
    print(f"Port:        {port}")
    print(f"Config:      {secretfile_path}")
    print(f"Docs:        http://{host}:{port}/docs")
    print(f"ReDoc:       http://{host}:{port}/redoc")
    print("━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n")

    # Check for API key
    if "SECRETZERO_API_KEY" in os.environ:
        print("🔑 API Key authentication enabled")
    else:
        print("⚠️  WARNING: Running in insecure mode (no API key required)")
        print("   Set SECRETZERO_API_KEY environment variable to enable authentication")

    print()

    # Run server
    uvicorn.run(
        app,
        host=host,
        port=port,
        reload=reload,
        log_level="info",
    )

Usage Examples

Creating a Custom App

from secretzero.api import create_app

# Create app with custom config path
app = create_app(secretfile_path="path/to/Secretfile.yml")

# Run with uvicorn
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)

Custom Middleware

from fastapi import FastAPI, Request
from secretzero.api import create_app

app = create_app()

@app.middleware("http")
async def custom_middleware(request: Request, call_next):
    # Add custom logic
    response = await call_next(request)
    return response

Custom Routes

from secretzero.api import create_app

app = create_app()

@app.get("/custom")
async def custom_endpoint():
    return {"message": "Custom endpoint"}

Testing

from fastapi.testclient import TestClient
from secretzero.api import create_app

app = create_app()
client = TestClient(app)

def test_health():
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"

Environment Variables

Variable	Description	Default
SECRETZERO_HOST	Host to bind to	0.0.0.0
SECRETZERO_PORT	Port to listen on	8000
SECRETZERO_CONFIG	Path to Secretfile	Secretfile.yml
SECRETZERO_API_KEY	API key for authentication	None (auth disabled)
SECRETZERO_RELOAD	Enable auto-reload	false
SECRETZERO_LOG_LEVEL	Logging level	info

Development

Running Locally

# Install with API extras
pip install -e ".[api,dev]"

# Set up environment
export SECRETZERO_API_KEY=$(python -c "import secrets; print(secrets.token_urlsafe(32))")
export SECRETZERO_CONFIG=Secretfile.yml

# Run with auto-reload
export SECRETZERO_RELOAD=true
secretzero-api

Testing the API

# Run API tests
pytest tests/test_api.py -v

# Run with coverage
pytest tests/test_api.py --cov=secretzero.api --cov-report=term-missing

Type Checking

# Check API types
mypy src/secretzero/api/

Architecture

Request Flow

graph LR
    A[Client Request] --> B[CORS Middleware]
    B --> C[Authentication]
    C --> D[Route Handler]
    D --> E[Business Logic]
    E --> F[Response]
    F --> G[Audit Log]
    G --> H[Client Response]

Components

• FastAPI App: Main application instance
• Middleware: CORS, authentication, error handling
• Routes: Endpoint definitions with validation
• Schemas: Pydantic models for request/response
• Business Logic: Core SecretZero operations
• Audit Logger: Track all API operations

Security

Authentication

The API uses API key authentication:

from secretzero.api.auth import RequireAuth

@app.get("/protected")
async def protected_endpoint(_auth: str = RequireAuth):
    return {"message": "Authenticated"}

CORS

Configure CORS for production:

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourdomain.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST"],
    allow_headers=["X-API-Key"],
)

Rate Limiting

Implement rate limiting with middleware:

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/endpoint")
@limiter.limit("5/minute")
async def limited_endpoint(request: Request):
    return {"message": "Rate limited"}

Performance

Caching

from functools import lru_cache

@lru_cache(maxsize=128)
def get_cached_config():
    return load_config()

Async Operations

All endpoints are async-capable:

@app.get("/async-operation")
async def async_endpoint():
    result = await async_operation()
    return result

Connection Pooling

Use connection pooling for external services:

import httpx

async_client = httpx.AsyncClient()

@app.on_event("shutdown")
async def shutdown():
    await async_client.aclose()

Deployment

Docker

FROM python:3.12-slim

WORKDIR /app
COPY . .
RUN pip install secretzero[api]

EXPOSE 8000
CMD ["secretzero-api"]

Docker Compose

version: '3.8'
services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - SECRETZERO_API_KEY=${API_KEY}
      - SECRETZERO_CONFIG=/app/Secretfile.yml
    volumes:
      - ./Secretfile.yml:/app/Secretfile.yml:ro

Kubernetes

apiVersion: apps/v1
kind: Deployment
metadata:
  name: secretzero-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: secretzero-api
  template:
    metadata:
      labels:
        app: secretzero-api
    spec:
      containers:
      - name: api
        image: secretzero:latest
        ports:
        - containerPort: 8000
        env:
        - name: SECRETZERO_API_KEY
          valueFrom:
            secretKeyRef:
              name: secretzero-api
              key: api-key

Reverse Proxy (nginx)

server {
    listen 80;
    server_name api.example.com;

    location / {
        proxy_pass http://localhost:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }
}

Next Steps

• Interactive API Reference - Test endpoints interactively
• API Getting Started - Quick start guide
• Examples - Real-world usage examples
• Configuration - Configure SecretZero