Add server certificate management UI (Settings > Endpoints > Server Certificate)
Operational Feature: Web UI for TLS certificate management
Problem:
- Server certificate (nginx HTTPS) managed manually via certs/ directory
- No visibility into cert expiry from UI
- Regeneration required command-line OpenSSL knowledge
- No warning when cert approaching expiration
- Agents using thumbprint pinning need new thumbprint after renewal
Solution:
- New admin-only settings page: Settings > Endpoints > Server Certificate
- View current certificate details (CN, SANs, expiry, thumbprint)
- Warning banner when cert expires within 30 days
- Generate self-signed certificate (configure CN, SANs, validity)
- Upload CA-signed certificate (PEM format)
- Automatic nginx restart after certificate changes
- Copy-to-clipboard for agent thumbprint pinning
Frontend:
- plugins/endpoints/frontend/views/ServerCertSettings.vue
- Certificate status display with expiry countdown
- Self-signed generation form (CN, SANs, days)
- CA certificate upload (cert + key files)
- Documentation collapsibles (requirements, pinning, air-gapped)
- plugins/endpoints/frontend/routes.js
- Route: /settings/endpoints/server-cert (admin-only)
Backend API:
- plugins/endpoints/api/routes.py
- GET /api/endpoints/settings/server-cert (view cert details)
- Uses openssl to extract CN, issuer, dates, thumbprint, SANs
- Calculates days until expiry (warns if < 30)
- Returns self-signed status
- POST /api/endpoints/settings/server-cert/generate
- Generates self-signed cert with custom CN/SANs/validity
- Backs up old certificate with timestamp
- Writes new cert/key to certs/ directory
- Restarts nginx (docker-compose restart nginx)
- Audit logged: endpoints.server_cert.generated
- POST /api/endpoints/settings/server-cert/upload
- Validates PEM format (cert + unencrypted key)
- Backs up old certificate
- Writes uploaded files to certs/
- Restarts nginx automatically
- Audit logged: endpoints.server_cert.uploaded
- All endpoints admin-only (@require_role('admin'))
Navigation:
- plugins/endpoints/plugin.py
- Added menu item: Settings > Endpoints > Server Certificate
- Icon: shield-check, Position: 29
Security:
- Admin-only access (certificate management is privileged operation)
- Private key validation (rejects encrypted keys)
- Certificate format validation (PEM-encoded)
- Audit logging for all cert operations
- Automatic backup before replacement (timestamped .backup files)
Operational Notes:
- Certificate changes restart nginx (brief HTTPS interruption)
- Agents using ServerCertThumbprint pinning must be updated after renewal
- Self-signed certs valid 1 year by default (configurable)
- Air-gapped sites: cert files in certs/ directory bundled automatically
Related:
- Complements certs/README.md and regeneration scripts (commit fd6240772e)
- Agents validate via Windows roots or thumbprint pinning (CertificateRenewal.cs)
- Client certificate renewal uses same nginx TLS surface (routes.py:3455)
Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -4281,3 +4281,342 @@ def acknowledge_fleet_alert(notification_id: int):
|
||||
message='Alert acknowledged', http_code=201)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Server certificate management
|
||||
# =============================================================================
|
||||
|
||||
@endpoints_bp.route('/settings/server-cert', methods=['GET'])
|
||||
@require_role('admin')
|
||||
def get_server_certificate():
|
||||
"""Get details about the current server certificate.
|
||||
|
||||
Returns certificate subject, expiry, thumbprint, and validation status.
|
||||
Admin-only - certificate details are sensitive operational data.
|
||||
"""
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
cert_path = Path('certs/server.crt')
|
||||
|
||||
if not cert_path.exists():
|
||||
return error_response(ErrorCodes.NOT_FOUND,
|
||||
'Server certificate not found at certs/server.crt',
|
||||
http_code=404)
|
||||
|
||||
try:
|
||||
# Extract certificate details using openssl
|
||||
subject_result = subprocess.run(
|
||||
['openssl', 'x509', '-in', str(cert_path), '-noout', '-subject'],
|
||||
capture_output=True, text=True, check=True
|
||||
)
|
||||
subject_cn = subject_result.stdout.strip().split('CN = ')[-1] if 'CN = ' in subject_result.stdout else 'Unknown'
|
||||
|
||||
issuer_result = subprocess.run(
|
||||
['openssl', 'x509', '-in', str(cert_path), '-noout', '-issuer'],
|
||||
capture_output=True, text=True, check=True
|
||||
)
|
||||
issuer = issuer_result.stdout.strip().split('issuer=')[-1] if 'issuer=' in issuer_result.stdout else 'Unknown'
|
||||
|
||||
dates_result = subprocess.run(
|
||||
['openssl', 'x509', '-in', str(cert_path), '-noout', '-dates'],
|
||||
capture_output=True, text=True, check=True
|
||||
)
|
||||
dates_lines = dates_result.stdout.strip().split('\n')
|
||||
not_before = dates_lines[0].split('notBefore=')[-1] if len(dates_lines) > 0 else None
|
||||
not_after = dates_lines[1].split('notAfter=')[-1] if len(dates_lines) > 1 else None
|
||||
|
||||
thumbprint_result = subprocess.run(
|
||||
['openssl', 'x509', '-in', str(cert_path), '-noout', '-fingerprint', '-sha256'],
|
||||
capture_output=True, text=True, check=True
|
||||
)
|
||||
thumbprint = thumbprint_result.stdout.strip().split('=')[-1] if '=' in thumbprint_result.stdout else 'Unknown'
|
||||
|
||||
# Get SANs
|
||||
san_result = subprocess.run(
|
||||
['openssl', 'x509', '-in', str(cert_path), '-noout', '-ext', 'subjectAltName'],
|
||||
capture_output=True, text=True
|
||||
)
|
||||
san_list = []
|
||||
if san_result.returncode == 0 and san_result.stdout:
|
||||
san_text = san_result.stdout.strip()
|
||||
if 'DNS:' in san_text or 'IP:' in san_text:
|
||||
san_parts = san_text.split('\n')
|
||||
for line in san_parts:
|
||||
if 'DNS:' in line or 'IP:' in line:
|
||||
items = line.strip().split(', ')
|
||||
san_list.extend([item.strip() for item in items if item.strip()])
|
||||
|
||||
# Calculate days until expiry
|
||||
if not_after:
|
||||
try:
|
||||
from dateutil import parser
|
||||
expiry_date = parser.parse(not_after)
|
||||
now = datetime.now(timezone.utc)
|
||||
if expiry_date.tzinfo is None:
|
||||
expiry_date = expiry_date.replace(tzinfo=timezone.utc)
|
||||
expires_in_days = (expiry_date - now).days
|
||||
except Exception:
|
||||
expires_in_days = None
|
||||
else:
|
||||
expires_in_days = None
|
||||
|
||||
# Check if self-signed
|
||||
is_self_signed = (subject_cn == issuer or 'CN = ' + subject_cn in issuer)
|
||||
|
||||
return success_response({
|
||||
'subject_cn': subject_cn,
|
||||
'issuer': issuer,
|
||||
'not_before': not_before,
|
||||
'not_after': not_after,
|
||||
'thumbprint': thumbprint,
|
||||
'san_list': san_list,
|
||||
'expires_in_days': expires_in_days,
|
||||
'is_self_signed': is_self_signed
|
||||
})
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
current_app.logger.error(f'Failed to read certificate: {e}')
|
||||
return error_response(ErrorCodes.INTERNAL_ERROR,
|
||||
f'Failed to read certificate details: {e.stderr}',
|
||||
http_code=500)
|
||||
except Exception as e:
|
||||
current_app.logger.error(f'Unexpected error reading certificate: {e}')
|
||||
return error_response(ErrorCodes.INTERNAL_ERROR,
|
||||
f'Failed to read certificate: {str(e)}',
|
||||
http_code=500)
|
||||
|
||||
|
||||
@endpoints_bp.route('/settings/server-cert/generate', methods=['POST'])
|
||||
@require_role('admin')
|
||||
@audit_log('endpoints.server_cert.generated')
|
||||
def generate_server_certificate():
|
||||
"""Generate a new self-signed server certificate.
|
||||
|
||||
Expects JSON body:
|
||||
{
|
||||
"cn": "localhost",
|
||||
"san": "DNS:localhost,IP:192.168.100.1",
|
||||
"days": 365
|
||||
}
|
||||
|
||||
Admin-only - generates new TLS certificate and restarts nginx.
|
||||
"""
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
|
||||
payload = request.get_json(silent=True) or {}
|
||||
|
||||
cn = (payload.get('cn') or '').strip()
|
||||
san = (payload.get('san') or '').strip()
|
||||
days = payload.get('days', 365)
|
||||
|
||||
if not cn:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'Common Name (cn) is required',
|
||||
http_code=400)
|
||||
|
||||
if not isinstance(days, int) or days < 1 or days > 3650:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'days must be an integer between 1 and 3650',
|
||||
http_code=400)
|
||||
|
||||
cert_dir = Path('certs')
|
||||
cert_path = cert_dir / 'server.crt'
|
||||
key_path = cert_dir / 'server.key'
|
||||
|
||||
# Backup existing certificate
|
||||
if cert_path.exists():
|
||||
backup_path = cert_dir / f'server.crt.backup.{int(time.time())}'
|
||||
try:
|
||||
cert_path.rename(backup_path)
|
||||
key_path.rename(cert_dir / f'server.key.backup.{int(time.time())}')
|
||||
current_app.logger.info(f'Backed up old certificate to {backup_path}')
|
||||
except Exception as e:
|
||||
current_app.logger.warning(f'Failed to backup old certificate: {e}')
|
||||
|
||||
# Create temporary OpenSSL config
|
||||
with tempfile.NamedTemporaryFile(mode='w', suffix='.cnf', delete=False) as config_file:
|
||||
config_content = f"""[req]
|
||||
distinguished_name = req_distinguished_name
|
||||
x509_extensions = v3_req
|
||||
prompt = no
|
||||
|
||||
[req_distinguished_name]
|
||||
CN = {cn}
|
||||
|
||||
[v3_req]
|
||||
subjectAltName = @alt_names
|
||||
|
||||
[alt_names]
|
||||
"""
|
||||
# Parse SANs and add to config
|
||||
if san:
|
||||
san_items = [s.strip() for s in san.split(',')]
|
||||
dns_count = 1
|
||||
ip_count = 1
|
||||
for item in san_items:
|
||||
if item.startswith('DNS:'):
|
||||
config_content += f'DNS.{dns_count} = {item[4:]}\n'
|
||||
dns_count += 1
|
||||
elif item.startswith('IP:'):
|
||||
config_content += f'IP.{ip_count} = {item[3:]}\n'
|
||||
ip_count += 1
|
||||
|
||||
config_file.write(config_content)
|
||||
config_path = config_file.name
|
||||
|
||||
try:
|
||||
# Generate certificate
|
||||
result = subprocess.run([
|
||||
'openssl', 'req', '-x509', '-newkey', 'rsa:2048', '-nodes',
|
||||
'-keyout', str(key_path),
|
||||
'-out', str(cert_path),
|
||||
'-days', str(days),
|
||||
'-config', config_path
|
||||
], capture_output=True, text=True, check=True)
|
||||
|
||||
current_app.logger.info(f'Generated new server certificate: CN={cn}, days={days}')
|
||||
|
||||
# Restart nginx to pick up new certificate
|
||||
try:
|
||||
restart_result = subprocess.run([
|
||||
'docker-compose', 'restart', 'nginx'
|
||||
], capture_output=True, text=True, timeout=30)
|
||||
|
||||
if restart_result.returncode == 0:
|
||||
current_app.logger.info('Nginx restarted successfully with new certificate')
|
||||
else:
|
||||
current_app.logger.warning(f'Nginx restart returned {restart_result.returncode}: {restart_result.stderr}')
|
||||
except Exception as restart_error:
|
||||
current_app.logger.error(f'Failed to restart nginx: {restart_error}')
|
||||
|
||||
return success_response({
|
||||
'message': 'Certificate generated successfully. Nginx is restarting.'
|
||||
})
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
current_app.logger.error(f'Failed to generate certificate: {e.stderr}')
|
||||
return error_response(ErrorCodes.INTERNAL_ERROR,
|
||||
f'Failed to generate certificate: {e.stderr}',
|
||||
http_code=500)
|
||||
except Exception as e:
|
||||
current_app.logger.error(f'Unexpected error generating certificate: {e}')
|
||||
return error_response(ErrorCodes.INTERNAL_ERROR,
|
||||
f'Failed to generate certificate: {str(e)}',
|
||||
http_code=500)
|
||||
finally:
|
||||
# Clean up temp config
|
||||
try:
|
||||
Path(config_path).unlink()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
@endpoints_bp.route('/settings/server-cert/upload', methods=['POST'])
|
||||
@require_role('admin')
|
||||
@audit_log('endpoints.server_cert.uploaded')
|
||||
def upload_server_certificate():
|
||||
"""Upload a new server certificate and private key.
|
||||
|
||||
Expects multipart/form-data with:
|
||||
- cert: Certificate file (.crt or .pem)
|
||||
- key: Private key file (.key or .pem)
|
||||
|
||||
Admin-only - replaces TLS certificate and restarts nginx.
|
||||
"""
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
if 'cert' not in request.files or 'key' not in request.files:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'Both cert and key files are required',
|
||||
http_code=400)
|
||||
|
||||
cert_file = request.files['cert']
|
||||
key_file = request.files['key']
|
||||
|
||||
if not cert_file.filename or not key_file.filename:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'Both cert and key files must be provided',
|
||||
http_code=400)
|
||||
|
||||
cert_dir = Path('certs')
|
||||
cert_path = cert_dir / 'server.crt'
|
||||
key_path = cert_dir / 'server.key'
|
||||
|
||||
# Validate certificate format before writing
|
||||
cert_content = cert_file.read().decode('utf-8')
|
||||
key_content = key_file.read().decode('utf-8')
|
||||
|
||||
if '-----BEGIN CERTIFICATE-----' not in cert_content:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'Invalid certificate format (must be PEM-encoded)',
|
||||
http_code=400)
|
||||
|
||||
if '-----BEGIN' not in key_content or 'PRIVATE KEY-----' not in key_content:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'Invalid private key format (must be PEM-encoded)',
|
||||
http_code=400)
|
||||
|
||||
# Check if private key is encrypted
|
||||
if 'ENCRYPTED' in key_content:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'Private key must be unencrypted (no passphrase)',
|
||||
http_code=400)
|
||||
|
||||
# Backup existing certificate
|
||||
if cert_path.exists():
|
||||
backup_path = cert_dir / f'server.crt.backup.{int(time.time())}'
|
||||
try:
|
||||
cert_path.rename(backup_path)
|
||||
key_path.rename(cert_dir / f'server.key.backup.{int(time.time())}')
|
||||
current_app.logger.info(f'Backed up old certificate to {backup_path}')
|
||||
except Exception as e:
|
||||
current_app.logger.warning(f'Failed to backup old certificate: {e}')
|
||||
|
||||
# Write new certificate and key
|
||||
try:
|
||||
cert_path.write_text(cert_content)
|
||||
key_path.write_text(key_content)
|
||||
|
||||
# Set restrictive permissions on private key
|
||||
key_path.chmod(0o600)
|
||||
|
||||
current_app.logger.info('Uploaded new server certificate and private key')
|
||||
|
||||
# Verify certificate is valid
|
||||
verify_result = subprocess.run([
|
||||
'openssl', 'x509', '-in', str(cert_path), '-noout', '-text'
|
||||
], capture_output=True, text=True)
|
||||
|
||||
if verify_result.returncode != 0:
|
||||
current_app.logger.error(f'Uploaded certificate is invalid: {verify_result.stderr}')
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR,
|
||||
'Uploaded certificate is not valid',
|
||||
http_code=400)
|
||||
|
||||
# Restart nginx to pick up new certificate
|
||||
try:
|
||||
restart_result = subprocess.run([
|
||||
'docker-compose', 'restart', 'nginx'
|
||||
], capture_output=True, text=True, timeout=30)
|
||||
|
||||
if restart_result.returncode == 0:
|
||||
current_app.logger.info('Nginx restarted successfully with uploaded certificate')
|
||||
else:
|
||||
current_app.logger.warning(f'Nginx restart returned {restart_result.returncode}: {restart_result.stderr}')
|
||||
except Exception as restart_error:
|
||||
current_app.logger.error(f'Failed to restart nginx: {restart_error}')
|
||||
|
||||
return success_response({
|
||||
'message': 'Certificate uploaded successfully. Nginx is restarting.'
|
||||
})
|
||||
|
||||
except Exception as e:
|
||||
current_app.logger.error(f'Failed to write certificate files: {e}')
|
||||
return error_response(ErrorCodes.INTERNAL_ERROR,
|
||||
f'Failed to write certificate files: {str(e)}',
|
||||
http_code=500)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user