154 lines
4.0 KiB
Python
154 lines
4.0 KiB
Python
"""Typed contracts for secure Agent enrollment."""
|
|
|
|
from datetime import datetime
|
|
from typing import Literal
|
|
from uuid import UUID
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
|
|
|
|
|
class AgentEnrollmentCreateRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
site_id: UUID
|
|
agent_zone_id: UUID
|
|
display_name: str = Field(min_length=1, max_length=120)
|
|
|
|
@field_validator("display_name")
|
|
@classmethod
|
|
def normalize_display_name(cls, value: str) -> str:
|
|
normalized = value.strip()
|
|
if not normalized:
|
|
raise ValueError("display_name cannot be blank")
|
|
return normalized
|
|
|
|
|
|
class AgentEnrollmentCreated(BaseModel):
|
|
id: UUID
|
|
site_id: UUID
|
|
site_name: str
|
|
agent_zone_id: UUID
|
|
agent_zone_name: str
|
|
display_name: str
|
|
created_at: datetime
|
|
expires_at: datetime
|
|
enrollment_token: str
|
|
|
|
|
|
class AgentEnrollmentCreateResponse(BaseModel):
|
|
data: AgentEnrollmentCreated
|
|
meta: dict[str, object] = Field(default_factory=dict)
|
|
error: None = None
|
|
|
|
|
|
class AgentEnrollmentSummary(BaseModel):
|
|
id: UUID
|
|
site_id: UUID
|
|
site_name: str
|
|
agent_zone_id: UUID
|
|
agent_zone_name: str
|
|
display_name: str
|
|
created_by_user_id: UUID | None = None
|
|
created_at: datetime
|
|
expires_at: datetime
|
|
used_at: datetime | None = None
|
|
revoked_at: datetime | None = None
|
|
target_agent_id: str | None = None
|
|
enrolled_agent_id: str | None = None
|
|
purpose: Literal["new_agent", "existing_agent_credential_transition"]
|
|
status: Literal["waiting", "used", "revoked", "expired"]
|
|
|
|
|
|
class AgentEnrollmentListResponse(BaseModel):
|
|
data: list[AgentEnrollmentSummary]
|
|
meta: dict[str, object] = Field(default_factory=dict)
|
|
error: None = None
|
|
|
|
|
|
class AgentEnrollRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
token: str = Field(min_length=32, max_length=256)
|
|
|
|
@field_validator("token")
|
|
@classmethod
|
|
def normalize_token(cls, value: str) -> str:
|
|
normalized = value.strip()
|
|
if not normalized:
|
|
raise ValueError("token cannot be blank")
|
|
return normalized
|
|
|
|
|
|
class AgentEnrollCredentials(BaseModel):
|
|
agent_id: str
|
|
agent_secret: str
|
|
site_id: UUID
|
|
site_name: str
|
|
agent_zone_id: UUID
|
|
agent_zone_name: str
|
|
display_name: str
|
|
ip_ranges: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class AgentEnrollResponse(BaseModel):
|
|
data: AgentEnrollCredentials
|
|
meta: dict[str, object] = Field(default_factory=dict)
|
|
error: None = None
|
|
|
|
|
|
class AgentWindowsPackageRequest(AgentEnrollRequest):
|
|
"""Short-lived token used to construct a Windows bootstrap package."""
|
|
|
|
|
|
class AgentCredentialTransitionCreateRequest(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
agent_id: str = Field(min_length=1, max_length=255)
|
|
|
|
@field_validator("agent_id")
|
|
@classmethod
|
|
def normalize_agent_id(cls, value: str) -> str:
|
|
normalized = value.strip()
|
|
if not normalized:
|
|
raise ValueError("agent_id cannot be blank")
|
|
return normalized
|
|
|
|
|
|
class AgentCredentialTransitionCreated(BaseModel):
|
|
id: UUID
|
|
target_agent_id: str
|
|
site_id: UUID
|
|
site_name: str
|
|
agent_zone_id: UUID
|
|
agent_zone_name: str
|
|
display_name: str
|
|
created_at: datetime
|
|
expires_at: datetime
|
|
transition_token: str
|
|
|
|
|
|
class AgentCredentialTransitionCreateResponse(BaseModel):
|
|
data: AgentCredentialTransitionCreated
|
|
meta: dict[str, object] = Field(default_factory=dict)
|
|
error: None = None
|
|
|
|
|
|
class AgentCredentialTransitionRedeemRequest(AgentEnrollRequest):
|
|
agent_secret_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
|
|
|
|
|
|
class AgentCredentialTransitionCredentials(BaseModel):
|
|
agent_id: str
|
|
site_id: UUID
|
|
site_name: str
|
|
agent_zone_id: UUID
|
|
agent_zone_name: str
|
|
display_name: str
|
|
ip_ranges: list[str] = Field(default_factory=list)
|
|
|
|
|
|
class AgentCredentialTransitionResponse(BaseModel):
|
|
data: AgentCredentialTransitionCredentials
|
|
meta: dict[str, object] = Field(default_factory=dict)
|
|
error: None = None
|