127 lines
3.6 KiB
Python
127 lines
3.6 KiB
Python
"""
|
|
/device endpoints.
|
|
"""
|
|
|
|
from datetime import datetime
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
|
|
from ..db import Connection, Device, DeviceCreate, DeviceRead, User
|
|
from ..easyrsa import CertificateType, DistinguishedName, EasyRSA
|
|
from ._common import (Responses, get_current_user, get_device_by_id,
|
|
get_user_by_name)
|
|
|
|
router = APIRouter(prefix="/device", tags=["device"])
|
|
|
|
|
|
@router.post(
|
|
"/{user_name}",
|
|
responses={
|
|
status.HTTP_201_CREATED: Responses.ENTRY_ADDED,
|
|
status.HTTP_400_BAD_REQUEST: Responses.NOT_INSTALLED,
|
|
status.HTTP_401_UNAUTHORIZED: Responses.NEEDS_USER,
|
|
status.HTTP_403_FORBIDDEN: Responses.NEEDS_PERMISSION,
|
|
status.HTTP_404_NOT_FOUND: Responses.ENTRY_DOESNT_EXIST,
|
|
status.HTTP_409_CONFLICT: Responses.ENTRY_EXISTS,
|
|
},
|
|
response_model=DeviceRead,
|
|
status_code=status.HTTP_201_CREATED,
|
|
)
|
|
async def add_device(
|
|
device: DeviceCreate,
|
|
current_user: User = Depends(get_current_user),
|
|
owner: User = Depends(get_user_by_name),
|
|
) -> Device:
|
|
"""
|
|
POST ./: Create a new device in the database.
|
|
"""
|
|
|
|
# check permission
|
|
if not current_user.can_create(Device, owner):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
|
|
|
|
# create the new device
|
|
new_device = Device.create(
|
|
owner=owner,
|
|
device=device,
|
|
)
|
|
|
|
# fail if creation was unsuccessful
|
|
if new_device is None:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT)
|
|
|
|
# return the created device on success
|
|
return new_device
|
|
|
|
|
|
@router.delete(
|
|
"/{device_id}",
|
|
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_PERMISSION,
|
|
},
|
|
response_model=User,
|
|
)
|
|
async def remove_device(
|
|
current_user: User = Depends(get_current_user),
|
|
device: Device = Depends(get_device_by_id),
|
|
):
|
|
"""
|
|
DELETE ./{device_id}: Remove a device from the database.
|
|
"""
|
|
|
|
# check permission
|
|
if not current_user.can_edit(device):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
|
|
|
|
# delete device
|
|
device.delete()
|
|
|
|
|
|
@router.post(
|
|
"/{device_id}/csr",
|
|
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_PERMISSION,
|
|
status.HTTP_404_NOT_FOUND: Responses.ENTRY_DOESNT_EXIST,
|
|
},
|
|
)
|
|
async def request_certificate(
|
|
current_user: User = Depends(get_current_user),
|
|
device: Device = Depends(get_device_by_id),
|
|
):
|
|
"""
|
|
POST ./{device_id}/csr: Request certificate for a device.
|
|
"""
|
|
|
|
# check permission
|
|
if not current_user.can_edit(device):
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
|
|
|
|
easy_rsa = EasyRSA()
|
|
|
|
with Connection.session as db:
|
|
db.add(device)
|
|
dn = DistinguishedName.build(device)
|
|
|
|
if current_user.can_issue(device):
|
|
device.approved = True
|
|
|
|
if (cert := easy_rsa.issue(
|
|
dn=dn,
|
|
cert_type=CertificateType.server,
|
|
)) is not None:
|
|
assert (expiry := cert.get_notAfter()) is not None
|
|
|
|
date_format, encoding = "%Y%m%d%H%M%SZ", "ascii"
|
|
expiry = datetime.strptime(
|
|
expiry.decode(encoding),
|
|
date_format,
|
|
)
|
|
device.expiry = expiry
|
|
|
|
db.commit()
|