coryHawkvelt/gatehouse-api
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Feat: Implemented SUDO Department & API Key, CA Serial

Browse Source
This commit is contained in:
James Bhattarai
2026-03-08 18:10:26 +05:45
parent ff976ee1cc
commit f334000da3
16 changed files with 911 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gatehouse_app/api/v1/__init__.py
+2 -1
View File
@@ -5,6 +5,7 @@ from flask import Blueprint
api_v1_bp = Blueprint("api_v1", __name__)
# Import route modules to register them
from gatehouse_app.api.v1 import auth, users, organizations, policies, external_auth, departments, principals, ssh, zerotier
from gatehouse_app.api.v1 import auth, users, organizations, policies, external_auth, departments, principals, ssh, zerotier, sudo
api_v1_bp.register_blueprint(ssh.ssh_bp)
gatehouse_app/api/v1/departments.py
+4
View File
@@ -16,12 +16,15 @@ class DepartmentCreateSchema(Schema):
"""Schema for creating a department."""
name = fields.Str(required=True, validate=validate.Length(min=1, max=255))
description = fields.Str(allow_none=True, validate=validate.Length(max=2000))
can_sudo = fields.Bool(allow_none=True, load_default=False)
class DepartmentUpdateSchema(Schema):
"""Schema for updating a department."""
name = fields.Str(validate=validate.Length(min=1, max=255))
description = fields.Str(allow_none=True, validate=validate.Length(max=2000))
can_sudo = fields.Bool(allow_none=True)
class AddDepartmentMemberSchema(Schema):
@@ -119,6 +122,7 @@ def create_department(org_id):
organization_id=org_id,
name=data["name"],
description=data.get("description"),
can_sudo=data.get("can_sudo", False),
)
db.session.add(dept)
db.session.commit()
gatehouse_app/api/v1/organizations/__init__.py
+2 -2
View File
@@ -1,4 +1,4 @@
"""Organization routes package."""
from gatehouse_app.api.v1.organizations import core, members, invites, clients, cas, audit, roles
from gatehouse_app.api.v1.organizations import core, members, invites, clients, cas, audit, roles, api_keys
__all__ = ["core", "members", "invites", "clients", "cas", "audit", "roles"]
__all__ = ["core", "members", "invites", "clients", "cas", "audit", "roles", "api_keys"]
gatehouse_app/api/v1/organizations/api_keys.py
+299
View File
@@ -0,0 +1,299 @@
"""Organization API Key management endpoints."""
from flask import g, request
from marshmallow import Schema, fields, validate, ValidationError
from gatehouse_app.api.v1 import api_v1_bp
from gatehouse_app.utils.response import api_response
from gatehouse_app.utils.decorators import login_required, require_admin, full_access_required
from gatehouse_app.models.organization import OrganizationApiKey
from gatehouse_app.services.organization_service import OrganizationService
from gatehouse_app.extensions import db
class ApiKeyCreateSchema(Schema):
"""Schema for creating an API key."""
name = fields.Str(required=True, validate=validate.Length(min=1, max=255))
description = fields.Str(allow_none=True, validate=validate.Length(max=2000))
class ApiKeyUpdateSchema(Schema):
"""Schema for updating an API key."""
name = fields.Str(validate=validate.Length(min=1, max=255))
description = fields.Str(allow_none=True, validate=validate.Length(max=2000))
@api_v1_bp.route("/organizations/<org_id>/api-keys", methods=["GET"])
@login_required
@require_admin
@full_access_required
def list_api_keys(org_id):
"""
List all API keys for an organization.
Only accessible by organization admins.
Args:
org_id: Organization ID
Returns:
200: List of API keys (without key values)
401: Not authenticated
403: Not an admin
404: Organization not found
"""
org = OrganizationService.get_organization_by_id(org_id)
# Check if user is an admin
from gatehouse_app.models.organization.organization_member import OrganizationMember
from gatehouse_app.utils.constants import OrganizationRole
membership = OrganizationMember.query.filter_by(
user_id=g.current_user.id,
organization_id=org_id,
deleted_at=None
).first()
if not membership or membership.role not in [OrganizationRole.OWNER, OrganizationRole.ADMIN]:
return api_response(
success=False,
message="You do not have permission to manage API keys",
status=403,
error_type="AUTHORIZATION_ERROR",
)
api_keys = OrganizationApiKey.query.filter_by(
organization_id=org_id,
deleted_at=None
).all()
return api_response(
data={
"api_keys": [k.to_dict() for k in api_keys],
"count": len(api_keys),
},
message="API keys retrieved successfully",
)
@api_v1_bp.route("/organizations/<org_id>/api-keys", methods=["POST"])
@login_required
@require_admin
@full_access_required
def create_api_key(org_id):
"""
Create a new API key for an organization.
Only accessible by organization admins.
The plain text key is returned only on creation and should be stored securely.
Args:
org_id: Organization ID
Request body:
name: API key name (required)
description: Optional description
Returns:
201: API key created successfully
400: Validation error
401: Not authenticated
403: Not an admin
404: Organization not found
"""
try:
org = OrganizationService.get_organization_by_id(org_id)
# Check if user is an admin
from gatehouse_app.models.organization.organization_member import OrganizationMember
from gatehouse_app.utils.constants import OrganizationRole
membership = OrganizationMember.query.filter_by(
user_id=g.current_user.id,
organization_id=org_id,
deleted_at=None
).first()
if not membership or membership.role not in [OrganizationRole.OWNER, OrganizationRole.ADMIN]:
return api_response(
success=False,
message="You do not have permission to create API keys",
status=403,
error_type="AUTHORIZATION_ERROR",
)
schema = ApiKeyCreateSchema()
data = schema.load(request.json or {})
# Create the API key
api_key, plain_key = OrganizationApiKey.create_key(
organization_id=org_id,
name=data["name"],
description=data.get("description"),
)
# Return the key data with the plain text key (only on creation)
key_dict = api_key.to_dict()
key_dict["key"] = plain_key # Include plain text only on creation
return api_response(
data={"api_key": key_dict},
message="API key created successfully. Store the key value securely - it cannot be retrieved later.",
status=201,
)
except ValidationError as e:
return api_response(
success=False,
message="Validation failed",
status=400,
error_type="VALIDATION_ERROR",
error_details=e.messages,
)
@api_v1_bp.route("/organizations/<org_id>/api-keys/<key_id>", methods=["PATCH"])
@login_required
@require_admin
@full_access_required
def update_api_key(org_id, key_id):
"""
Update an API key.
Only accessible by organization admins.
Args:
org_id: Organization ID
key_id: API Key ID
Request body:
name: New name (optional)
description: New description (optional)
Returns:
200: API key updated successfully
400: Validation error
401: Not authenticated
403: Not an admin
404: Organization or API key not found
"""
try:
org = OrganizationService.get_organization_by_id(org_id)
# Check if user is an admin
from gatehouse_app.models.organization.organization_member import OrganizationMember
from gatehouse_app.utils.constants import OrganizationRole
membership = OrganizationMember.query.filter_by(
user_id=g.current_user.id,
organization_id=org_id,
deleted_at=None
).first()
if not membership or membership.role not in [OrganizationRole.OWNER, OrganizationRole.ADMIN]:
return api_response(
success=False,
message="You do not have permission to update API keys",
status=403,
error_type="AUTHORIZATION_ERROR",
)
api_key = OrganizationApiKey.query.filter_by(
id=key_id,
organization_id=org_id,
deleted_at=None
).first()
if not api_key:
return api_response(
success=False,
message="API key not found",
status=404,
error_type="NOT_FOUND",
)
schema = ApiKeyUpdateSchema()
data = schema.load(request.json or {})
# Update fields
if "name" in data:
api_key.name = data["name"]
if "description" in data:
api_key.description = data["description"]
api_key.save()
return api_response(
data={"api_key": api_key.to_dict()},
message="API key updated successfully",
)
except ValidationError as e:
return api_response(
success=False,
message="Validation failed",
status=400,
error_type="VALIDATION_ERROR",
error_details=e.messages,
)
@api_v1_bp.route("/organizations/<org_id>/api-keys/<key_id>", methods=["DELETE"])
@login_required
@require_admin
@full_access_required
def delete_api_key(org_id, key_id):
"""
Delete/revoke an API key.
Only accessible by organization admins.
Args:
org_id: Organization ID
key_id: API Key ID
Returns:
200: API key deleted successfully
401: Not authenticated
403: Not an admin
404: Organization or API key not found
"""
org = OrganizationService.get_organization_by_id(org_id)
# Check if user is an admin
from gatehouse_app.models.organization.organization_member import OrganizationMember
from gatehouse_app.utils.constants import OrganizationRole
membership = OrganizationMember.query.filter_by(
user_id=g.current_user.id,
organization_id=org_id,
deleted_at=None
).first()
if not membership or membership.role not in [OrganizationRole.OWNER, OrganizationRole.ADMIN]:
return api_response(
success=False,
message="You do not have permission to delete API keys",
status=403,
error_type="AUTHORIZATION_ERROR",
)
api_key = OrganizationApiKey.query.filter_by(
id=key_id,
organization_id=org_id,
deleted_at=None
).first()
if not api_key:
return api_response(
success=False,
message="API key not found",
status=404,
error_type="NOT_FOUND",
)
# Soft delete the API key
api_key.delete(soft=True)
return api_response(
message="API key deleted successfully",
)
gatehouse_app/api/v1/organizations/audit.py
+98
View File
@@ -173,3 +173,101 @@ def get_my_audit_logs():
},
message="Activity retrieved",
)
@api_v1_bp.route("/organizations/<org_id>/certificates/audit", methods=["GET"])
@login_required
@require_admin
@full_access_required
def get_certificate_audit_logs(org_id):
"""
Get certificate issuance audit logs for an organization.
Only accessible by organization admins.
Returns certificate serial IDs, user IDs, and issuance timestamps for compliance.
Args:
org_id: Organization ID
Returns:
200: List of certificate audit logs
401: Not authenticated
403: Not an admin
404: Organization not found
"""
from gatehouse_app.models.ssh_ca.certificate_audit_log import CertificateAuditLog
from gatehouse_app.models.ssh_ca.ssh_certificate import SSHCertificate
from gatehouse_app.models.user import User
org = OrganizationService.get_organization_by_id(org_id)
page = max(1, int(request.args.get("page", 1)))
per_page = min(int(request.args.get("per_page", 50)), 200)
action_filter = request.args.get("action", "signed") # Default to signed certificates
# Get all CAs for this organization
from gatehouse_app.models.ssh_ca import CA
org_cas = CA.query.filter_by(organization_id=org_id, deleted_at=None).all()
org_ca_ids = [ca.id for ca in org_cas]
if not org_ca_ids:
return api_response(
data={
"audit_logs": [],
"count": 0,
"page": page,
"per_page": per_page,
"pages": 0,
},
message="No certificate audit logs found",
)
# Query certificate audit logs for certificates issued by org's CAs
query = CertificateAuditLog.query.join(
SSHCertificate,
CertificateAuditLog.certificate_id == SSHCertificate.id
).filter(
SSHCertificate.ca_id.in_(org_ca_ids),
CertificateAuditLog.deleted_at.is_(None)
)
if action_filter:
query = query.filter(CertificateAuditLog.action == action_filter)
query = query.order_by(CertificateAuditLog.created_at.desc())
total = query.count()
logs = query.offset((page - 1) * per_page).limit(per_page).all()
# Build response data with certificate details
audit_data = []
for log in logs:
cert = log.certificate
user = log.user
audit_data.append({
"id": log.id,
"action": log.action,
"certificate_serial": cert.serial,
"key_id": cert.key_id,
"principals": cert.principals,
"user_id": user.id if user else cert.user_id,
"user_email": user.email if user else None,
"issued_at": cert.created_at.isoformat() if cert.created_at else None,
"valid_after": cert.valid_after.isoformat() if cert.valid_after else None,
"valid_before": cert.valid_before.isoformat() if cert.valid_before else None,
"ip_address": log.ip_address,
"user_agent": log.user_agent,
"message": log.message,
"success": log.success,
"created_at": log.created_at.isoformat() if log.created_at else None,
})
return api_response(
data={
"audit_logs": audit_data,
"count": total,
"page": page,
"per_page": per_page,
"pages": (total + per_page - 1) // per_page,
},
message="Certificate audit logs retrieved successfully",
)
gatehouse_app/api/v1/sudo.py
+137
View File
@@ -0,0 +1,137 @@
"""Sudoer check and sudo-related endpoints."""
from flask import request
from gatehouse_app.api.v1 import api_v1_bp
from gatehouse_app.utils.response import api_response
from gatehouse_app.models.organization import OrganizationApiKey
from gatehouse_app.models.ssh_ca.ssh_certificate import SSHCertificate
from gatehouse_app.models.organization import Department, DepartmentMembership
@api_v1_bp.route("/sudo/check", methods=["POST"])
def check_sudoer():
"""
Check if a user with a given certificate can sudo.
This endpoint validates an API key for an organization, retrieves the certificate
by serial ID, finds the user and their departments, and checks if any of their
departments have sudo capability.
Request body:
api_key: Organization API key (required)
certificate_serial: Certificate serial ID (required)
Returns:
200: Sudoer status returned
400: Invalid request body
401: Invalid API key
403: Certificate not found or user not found
404: Organization or certificate not found
"""
try:
data = request.get_json()
if not data:
return api_response(
success=False,
message="Request body is required",
status=400,
error_type="INVALID_REQUEST",
)
api_key = data.get("api_key")
certificate_serial = data.get("certificate_serial")
if not api_key or certificate_serial is None:
return api_response(
success=False,
message="api_key and certificate_serial are required",
status=400,
error_type="MISSING_REQUIRED_FIELDS",
)
# Find the certificate by serial
certificate = SSHCertificate.query.filter_by(
serial=certificate_serial,
deleted_at=None
).first()
if not certificate:
return api_response(
success=False,
message="Certificate not found",
status=404,
error_type="NOT_FOUND",
)
# Get the CA and organization
ca = certificate.ca
if not ca:
return api_response(
success=False,
message="Certificate CA not found",
status=404,
error_type="NOT_FOUND",
)
org_id = ca.organization_id
# Verify the API key for this organization
org_api_key = OrganizationApiKey.verify_key(org_id, api_key)
if not org_api_key:
return api_response(
success=False,
message="Invalid API key for organization",
status=401,
error_type="UNAUTHORIZED",
)
# Get the user from the certificate
user = certificate.user
if not user:
return api_response(
success=False,
message="Certificate user not found",
status=404,
error_type="NOT_FOUND",
)
# Get all departments the user belongs to
user_departments = DepartmentMembership.query.filter_by(
user_id=user.id,
deleted_at=None
).all()
# Check if any of the user's departments have sudo capability
can_sudo = False
sudoer_departments = []
for dept_membership in user_departments:
dept = dept_membership.department
if dept and dept.can_sudo and dept.deleted_at is None:
can_sudo = True
sudoer_departments.append({
"id": dept.id,
"name": dept.name,
})
return api_response(
data={
"can_sudo": can_sudo,
"user_id": user.id,
"user_email": user.email,
"certificate_serial": certificate.serial,
"sudoer_departments": sudoer_departments,
"all_departments_count": len(user_departments),
},
message="Sudoer status retrieved successfully",
status=200,
)
except Exception as e:
return api_response(
success=False,
message=f"An error occurred: {str(e)}",
status=500,
error_type="INTERNAL_ERROR",
)