Endpoint POST api/dn

api/kiwi_vpn_api/db/schemas.py

@@ -10,7 +10,7 @@ from enum import Enum
 from typing import Any
 
 from passlib.context import CryptContext
-from pydantic import BaseModel, validator
+from pydantic import BaseModel, Field, constr, validator
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import Session
 
@@ -23,12 +23,12 @@ from . import models
 
 class DistinguishedNameBase(BaseModel):
     cn_only: bool
-    country: str
-    state: str
-    city: str
-    organization: str
-    organizational_unit: str
-    email: str
+    country: constr(max_length=2) | None
+    state: str | None
+    city: str | None
+    organization: str | None
+    organizational_unit: str | None
+    email: str | None
     common_name: str
 
 
@@ -40,6 +40,45 @@ class DistinguishedName(DistinguishedNameBase):
     class Config:
         orm_mode = True
 
+    @classmethod
+    def create(
+        cls,
+        db: Session,
+        dn: DistinguishedNameCreate,
+        owner: User,
+    ) -> User | None:
+        """
+        Create a new distinguished name in the database.
+        """
+
+        try:
+            db_owner = models.User.load(
+                db=db,
+                name=owner.name,
+            )
+
+            dn = models.DistinguishedName(
+                cn_only=dn.cn_only,
+                country=dn.country,
+                state=dn.state,
+                city=dn.city,
+                organization=dn.organization,
+                organizational_unit=dn.organizational_unit,
+                email=dn.email,
+                common_name=dn.common_name,
+                owner=db_owner,
+            )
+
+            db.add(dn)
+            db.commit()
+            db.refresh(dn)
+
+            return cls.from_orm(dn)
+
+        except IntegrityError:
+            # distinguished name already existed
+            pass
+
 ##########
 # table: certificates
 ##########
@@ -67,6 +106,9 @@ class Certificate(CertificateBase):
 class UserCapability(Enum):
     admin = "admin"
 
+    def __repr__(self) -> str:
+        return self.value
+
     @classmethod
     def from_value(cls, value) -> UserCapability:
         """
@@ -100,8 +142,14 @@ class UserCreate(UserBase):
 
 class User(UserBase):
     capabilities: list[UserCapability] = []
-    distinguished_names: list[DistinguishedName] = []
-    certificates: list[Certificate] = []
+
+    distinguished_names: list[DistinguishedName] = Field(
+        default=[], repr=False
+    )
+
+    certificates: list[Certificate] = Field(
+        default=[], repr=False
+    )
 
     class Config:
         orm_mode = True

api/kiwi_vpn_api/main.py

@@ -15,7 +15,7 @@ from fastapi import FastAPI
 from .config import Config, Settings
 from .db import Connection
 from .db.schemas import User
-from .routers import admin, user
+from .routers import admin, dn, user
 
 settings = Settings.get()
 
@@ -39,6 +39,7 @@ api = FastAPI(
 
 api.include_router(admin.router)
 api.include_router(user.router)
+api.include_router(dn.router)
 
 app.mount("/api", api)
 

api/kiwi_vpn_api/routers/_common.py

@@ -40,6 +40,10 @@ class Responses:
         "description": "Must be admin",
         "content": None,
     }
+    NEEDS_ADMIN_OR_SELF = {
+        "description": "Must be the requested user",
+        "content": None,
+    }
     ENTRY_EXISTS = {
         "description": "Entry exists in database",
         "content": None,

api/kiwi_vpn_api/routers/dn.py

@@ -0,0 +1,58 @@
+"""
+/dn endpoints.
+"""
+
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+
+from ..db import Connection
+from ..db.schemas import DistinguishedName, DistinguishedNameCreate, User
+from ._common import Responses, get_current_user_if_admin_or_self
+
+router = APIRouter(prefix="/dn")
+
+
+@router.post(
+    "",
+    responses={
+        status.HTTP_200_OK: Responses.OK,
+        status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
+        status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
+        status.HTTP_403_FORBIDDEN: Responses.NEEDS_ADMIN,
+        status.HTTP_404_NOT_FOUND: Responses.ENTRY_DOESNT_EXIST,
+        status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
+    },
+)
+async def add_distinguished_name(
+    user_name: str,
+    distinguished_name: DistinguishedNameCreate,
+    _: User = Depends(get_current_user_if_admin_or_self),
+    db: Session | None = Depends(Connection.get),
+):
+    """
+    POST ./: Create a new distinguished name in the database.
+    """
+
+    owner = User.from_db(
+        db=db,
+        name=user_name,
+    )
+
+    # fail if owner doesn't exist
+    if owner is None:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT)
+
+    # actually create the new user
+    new_dn = DistinguishedName.create(
+        db=db,
+        dn=distinguished_name,
+        owner=owner,
+    )
+
+    # fail if creation was unsuccessful
+    if new_dn is None:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT)
+
+    # return the created user on success
+    return new_dn