Add team webhooks and deploy tokens

app/forms/project.py

@@ -587,6 +587,25 @@ def validate_confirm(self, field):
             raise ValidationError(_("Project name confirmation did not match."))
 
 
+class ProjectWebhooksForm(StarletteForm):
+    webhook_url = StringField(
+        _l("Webhook URL"),
+        validators=[
+            Optional(),
+            Length(max=2048),
+            Regexp(
+                r"^https?://",
+                message=_l("URL must start with http:// or https://"),
+            ),
+        ],
+    )
+    webhook_secret = StringField(
+        _l("Secret"),
+        validators=[Optional(), Length(max=255)],
+    )
+    webhook_events = FieldList(StringField(), min_entries=0)
+
+
 class ProjectDeployForm(StarletteForm):
     commit = HiddenField(_l("Commit"), validators=[DataRequired()])
     submit = SubmitField(_l("Deploy"))
@@ -605,3 +624,33 @@ class ProjectPromoteDeploymentForm(StarletteForm):
     environment_id = HiddenField(_l("Environment ID"), validators=[DataRequired()])
     deployment_id = HiddenField(_l("Deployment ID"), validators=[DataRequired()])
     submit = SubmitField(_l("Promote"))
+
+
+class DeployTokenForm(StarletteForm):
+    token_id = HiddenField()
+    name = StringField(
+        _l("Name"),
+        validators=[DataRequired(), Length(min=1, max=100)],
+    )
+    environment_id = SelectField(_l("Environment"), choices=[])
+
+    def __init__(self, *args, project: Project | None = None, **kwargs):
+        super().__init__(*args, **kwargs)
+        if project:
+            self.environment_id.choices = [("", _l("All environments"))] + [
+                (env["id"], env["name"]) for env in project.active_environments
+            ]
+
+
+class DeleteDeployTokenForm(StarletteForm):
+    token_id = HiddenField(validators=[DataRequired()])
+    confirm = StringField(_l("Confirmation"), validators=[DataRequired()])
+    submit = SubmitField(_l("Revoke"))
+
+    def __init__(self, *args, token_name: str | None = None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.token_name = token_name
+
+    def validate_confirm(self, field):
+        if self.token_name and field.data != self.token_name:
+            raise ValidationError(_("Token name confirmation did not match."))

app/forms/team.py

@@ -6,8 +6,17 @@
     FileField,
     HiddenField,
     SelectField,
+    SelectMultipleField,
+    FieldList,
+)
+from wtforms.validators import (
+    ValidationError,
+    DataRequired,
+    Length,
+    Regexp,
+    Email,
+    Optional,
 )
-from wtforms.validators import ValidationError, DataRequired, Length, Regexp, Email
 from sqlalchemy import select, func
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -226,3 +235,50 @@ class TeamLeaveForm(StarletteForm):
 class TeamInviteAcceptForm(StarletteForm):
     invite_id = HiddenField(validators=[DataRequired()])
     submit = SubmitField(_l("Accept"))
+
+
+WEBHOOK_EVENTS = [
+    ("started", _l("Deployment Started")),
+    ("succeeded", _l("Deployment Succeeded")),
+    ("failed", _l("Deployment Failed")),
+    ("canceled", _l("Deployment Canceled")),
+]
+
+
+class TeamWebhookForm(StarletteForm):
+    webhook_id = HiddenField()
+    name = StringField(
+        _l("Name"),
+        validators=[DataRequired(), Length(min=1, max=100)],
+    )
+    url = StringField(
+        _l("URL"),
+        validators=[
+            DataRequired(),
+            Length(max=2048),
+            Regexp(
+                r"^https?://",
+                message=_l("URL must start with http:// or https://"),
+            ),
+        ],
+    )
+    secret = StringField(
+        _l("Secret"),
+        validators=[Optional(), Length(max=255)],
+    )
+    events = FieldList(StringField(), min_entries=0)
+    project_ids = FieldList(StringField(), min_entries=0)
+
+
+class TeamDeleteWebhookForm(StarletteForm):
+    webhook_id = HiddenField(validators=[DataRequired()])
+    confirm = StringField(_l("Confirmation"), validators=[DataRequired()])
+    submit = SubmitField(_l("Delete"))
+
+    def __init__(self, *args, webhook_name: str | None = None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.webhook_name = webhook_name
+
+    def validate_confirm(self, field):
+        if self.webhook_name and field.data != self.webhook_name:
+            raise ValidationError(_("Webhook name confirmation did not match."))

app/main.py

@@ -17,7 +17,7 @@
 from db import get_db, AsyncSessionLocal
 from dependencies import get_current_user, TemplateResponse
 from models import User, Team, Deployment, Project
-from routers import auth, project, github, google, team, user, event, admin
+from routers import auth, project, github, google, team, user, event, admin, api
 from services.loki import LokiService
 
 settings = get_settings()
@@ -158,6 +158,7 @@ async def root(
 app.include_router(google.router)
 app.include_router(team.router)
 app.include_router(event.router)
+app.include_router(api.router)
 
 
 @app.exception_handler(404)

app/migrations/versions/c1a2b3d4e5f6_webhooks_and_deploy_tokens.py

@@ -0,0 +1,108 @@
+"""webhooks_and_deploy_tokens
+
+Revision ID: c1a2b3d4e5f6
+Revises: 87a893d57c86
+Create Date: 2025-12-30 12:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "c1a2b3d4e5f6"
+down_revision: Union[str, Sequence[str], None] = "87a893d57c86"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # Create team_webhook table
+    op.create_table(
+        "team_webhook",
+        sa.Column("id", sa.String(length=32), nullable=False),
+        sa.Column("team_id", sa.String(length=32), nullable=False),
+        sa.Column("name", sa.String(length=100), nullable=False),
+        sa.Column("url", sa.String(length=2048), nullable=False),
+        sa.Column("secret", sa.String(length=512), nullable=True),
+        sa.Column("events", sa.JSON(), nullable=False),
+        sa.Column("project_ids", sa.JSON(), nullable=True),
+        sa.Column(
+            "status",
+            sa.Enum("active", "disabled", name="team_webhook_status"),
+            nullable=False,
+        ),
+        sa.Column("created_by_user_id", sa.Integer(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.Column("updated_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["team_id"],
+            ["team.id"],
+        ),
+        sa.ForeignKeyConstraint(
+            ["created_by_user_id"], ["user.id"], ondelete="SET NULL", use_alter=True
+        ),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_team_webhook_team_id"), "team_webhook", ["team_id"], unique=False
+    )
+    op.create_index(
+        op.f("ix_team_webhook_created_at"), "team_webhook", ["created_at"], unique=False
+    )
+    op.create_index(
+        op.f("ix_team_webhook_updated_at"), "team_webhook", ["updated_at"], unique=False
+    )
+
+    # Create deploy_token table
+    op.create_table(
+        "deploy_token",
+        sa.Column("id", sa.String(length=32), nullable=False),
+        sa.Column("project_id", sa.String(length=32), nullable=False),
+        sa.Column("name", sa.String(length=100), nullable=False),
+        sa.Column("token", sa.String(length=128), nullable=False),
+        sa.Column("environment_id", sa.String(length=8), nullable=True),
+        sa.Column(
+            "status",
+            sa.Enum("active", "revoked", name="deploy_token_status"),
+            nullable=False,
+        ),
+        sa.Column("last_used_at", sa.DateTime(), nullable=True),
+        sa.Column("created_by_user_id", sa.Integer(), nullable=True),
+        sa.Column("created_at", sa.DateTime(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["project_id"],
+            ["project.id"],
+        ),
+        sa.ForeignKeyConstraint(
+            ["created_by_user_id"], ["user.id"], ondelete="SET NULL", use_alter=True
+        ),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("token"),
+    )
+    op.create_index(
+        op.f("ix_deploy_token_project_id"), "deploy_token", ["project_id"], unique=False
+    )
+    op.create_index(
+        op.f("ix_deploy_token_created_at"), "deploy_token", ["created_at"], unique=False
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # Drop deploy_token table
+    op.drop_index(op.f("ix_deploy_token_created_at"), table_name="deploy_token")
+    op.drop_index(op.f("ix_deploy_token_project_id"), table_name="deploy_token")
+    op.drop_table("deploy_token")
+    op.execute("DROP TYPE IF EXISTS deploy_token_status")
+
+    # Drop team_webhook table
+    op.drop_index(op.f("ix_team_webhook_updated_at"), table_name="team_webhook")
+    op.drop_index(op.f("ix_team_webhook_created_at"), table_name="team_webhook")
+    op.drop_index(op.f("ix_team_webhook_team_id"), table_name="team_webhook")
+    op.drop_table("team_webhook")
+    op.execute("DROP TYPE IF EXISTS team_webhook_status")

app/models.py

@@ -849,3 +849,135 @@ class Allowlist(Base):
     @override
     def __repr__(self):
         return f"<Allowlist {self.type}:{self.value}>"
+
+
+class TeamWebhook(Base):
+    """Team-level webhook configuration for deployment events.
+
+    Webhooks can be scoped to specific projects or apply to all projects in the team.
+    """
+
+    __tablename__: str = "team_webhook"
+
+    id: Mapped[str] = mapped_column(
+        String(32), primary_key=True, default=lambda: token_hex(16)
+    )
+    team_id: Mapped[str] = mapped_column(ForeignKey("team.id"), index=True)
+    name: Mapped[str] = mapped_column(String(100), nullable=False)
+    url: Mapped[str] = mapped_column(String(2048), nullable=False)
+    _secret: Mapped[str | None] = mapped_column("secret", String(512), nullable=True)
+    events: Mapped[list[str]] = mapped_column(JSON, nullable=False, default=list)
+    project_ids: Mapped[list[str] | None] = mapped_column(
+        JSON, nullable=True
+    )  # None = all projects
+    status: Mapped[str] = mapped_column(
+        SQLAEnum("active", "disabled", name="team_webhook_status"),
+        nullable=False,
+        default="active",
+    )
+    created_by_user_id: Mapped[int | None] = mapped_column(
+        ForeignKey("user.id", use_alter=True, ondelete="SET NULL"), nullable=True
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        index=True, nullable=False, default=utc_now
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        index=True, nullable=False, default=utc_now, onupdate=utc_now
+    )
+
+    # Relationships
+    team: Mapped[Team] = relationship()
+    created_by_user: Mapped[User | None] = relationship(
+        foreign_keys=[created_by_user_id]
+    )
+
+    @property
+    def secret(self) -> str | None:
+        if self._secret:
+            fernet = get_fernet()
+            return fernet.decrypt(self._secret.encode()).decode()
+        return None
+
+    @secret.setter
+    def secret(self, value: str | None):
+        if value:
+            fernet = get_fernet()
+            self._secret = fernet.encrypt(value.encode()).decode()
+        else:
+            self._secret = None
+
+    def applies_to_project(self, project_id: str) -> bool:
+        """Check if this webhook applies to the given project."""
+        if self.project_ids is None:
+            return True  # All projects
+        return project_id in self.project_ids
+
+    @override
+    def __repr__(self):
+        return f"<TeamWebhook {self.name}>"
+
+
+class DeployToken(Base):
+    """Deploy token for triggering deployments via API/webhook.
+
+    Tokens can be scoped to specific environments or all environments.
+    """
+
+    __tablename__: str = "deploy_token"
+
+    id: Mapped[str] = mapped_column(
+        String(32), primary_key=True, default=lambda: token_hex(16)
+    )
+    project_id: Mapped[str] = mapped_column(ForeignKey("project.id"), index=True)
+    name: Mapped[str] = mapped_column(String(100), nullable=False)
+    _token: Mapped[str] = mapped_column(
+        "token", String(128), nullable=False, unique=True
+    )
+    environment_id: Mapped[str | None] = mapped_column(
+        String(8), nullable=True
+    )  # None = all environments
+    status: Mapped[str] = mapped_column(
+        SQLAEnum("active", "revoked", name="deploy_token_status"),
+        nullable=False,
+        default="active",
+    )
+    last_used_at: Mapped[datetime | None] = mapped_column(nullable=True)
+    created_by_user_id: Mapped[int | None] = mapped_column(
+        ForeignKey("user.id", use_alter=True, ondelete="SET NULL"), nullable=True
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        index=True, nullable=False, default=utc_now
+    )
+
+    # Relationships
+    project: Mapped[Project] = relationship()
+    created_by_user: Mapped[User | None] = relationship(
+        foreign_keys=[created_by_user_id]
+    )
+
+    @classmethod
+    def generate_token(cls) -> str:
+        """Generate a secure random token."""
+        return f"dp_{token_hex(32)}"
+
+    @property
+    def token(self) -> str:
+        """Get the token (stored as hash, so this returns the hash)."""
+        return self._token
+
+    @classmethod
+    def hash_token(cls, raw_token: str) -> str:
+        """Hash a raw token for storage."""
+        import hashlib
+
+        return hashlib.sha256(raw_token.encode()).hexdigest()
+
+    def can_deploy_environment(self, environment_id: str) -> bool:
+        """Check if this token can deploy to the given environment."""
+        if self.environment_id is None:
+            return True  # All environments
+        return self.environment_id == environment_id
+
+    @override
+    def __repr__(self):
+        return f"<DeployToken {self.name}>"