Merge branch 'tmp/sec2' into feat/ui-security-polish

This commit is contained in:
Juan Denis committed 2026-11-03 13:40:02 -05:00
commit a44e3c2435
32 files changed
+602 -96

No files matched your search

+7
View File
@@ -137,6 +137,13 @@ carry a `sink-safe: <sanitizer> — <why>` comment. `scripts/check-html-sinks.mj
fails CI on any new unannotated sink, so a raw-HTML injection point can't land
silently.
## Known Limitations
- **Tokens in `localStorage`.** The SPA keeps its JWT access/refresh tokens in
`localStorage`, so an XSS bug would expose them. The production CSP now
forbids inline and `eval` scripts (`script-src 'self'`) to reduce that risk;
planned follow-up: move tokens to `httpOnly` `SameSite` cookies with CSRF protection.
## Extension Trust Model
Extensions are **in-process code**. An installed extension runs with the panel's
+6 -2
View File
@@ -1,3 +1,4 @@
import logging
from flask import Blueprint, request, jsonify
from flask_jwt_extended import jwt_required
from app.services.advanced_ssl_service import AdvancedSSLService
@@ -6,6 +7,8 @@ from app.models.audit_log import AuditLog
from app.middleware.rbac import get_current_user
from app.api._request import json_object
logger = logging.getLogger(__name__)
advanced_ssl_bp = Blueprint('advanced_ssl', __name__)
@@ -79,8 +82,9 @@ def upload_cert():
try:
result = AdvancedSSLService.upload_custom_cert(domain, cert, key, chain)
return jsonify(result), 201
except Exception as e:
return jsonify({'error': str(e)}), 400
except Exception:
logger.exception('Custom certificate upload failed')
return jsonify({'error': 'Could not install the certificate'}), 400
@advanced_ssl_bp.route('/health/<domain>', methods=['GET'])
+27 -3
View File
@@ -1,3 +1,4 @@
import logging
import os
from flask import Blueprint, request, jsonify
from flask_jwt_extended import jwt_required
@@ -7,10 +8,21 @@ from app.services.backup_service import BackupService
from app.services.backup_policy_service import BackupPolicyService, BackupPolicyError
from app.services.storage_provider_service import StorageProviderService
from app import db, paths
from app.exceptions import field_required, not_found
from app.exceptions import DependencyUnavailableError, field_required, not_found
from app.api._request import json_object
backups_bp = Blueprint('backups', __name__)
logger = logging.getLogger(__name__)
_STORAGE_UNAVAILABLE = ('Backup storage is not accessible: the panel cannot create or read '
'its backup directory. Check its ownership and permissions.')
@backups_bp.errorhandler(PermissionError)
def _backup_dir_unavailable(exc):
logger.error('Backup directory is not writable: %s', exc)
err = DependencyUnavailableError('Backup directory is not writable', code='backup_dir_unavailable')
return jsonify(err.to_dict()), err.status_code
# UI-facing metadata for the CORE target types. Extension kinds
@@ -62,7 +74,12 @@ def list_target_types():
def list_backups():
"""List all backups."""
backup_type = request.args.get('type')
backups = BackupService.list_backups(backup_type)
try:
backups = BackupService.list_backups(backup_type)
except OSError:
# e.g. PermissionError creating /var/backups/serverkit: keep the page usable.
logger.exception('Backup directory is not accessible')
return jsonify({'backups': [], 'warning': _STORAGE_UNAVAILABLE}), 200
return jsonify({'backups': backups}), 200
@@ -70,7 +87,14 @@ def list_backups():
@admin_required
def get_stats():
"""Get backup statistics."""
stats = BackupService.get_backup_stats()
try:
stats = BackupService.get_backup_stats()
except OSError:
logger.exception('Backup directory is not accessible')
stats = {'total_backups': 0, 'application_backups': 0, 'database_backups': 0,
'file_backups': 0, 'total_size': 0, 'total_size_human': '0 B',
'remote_count': 0, 'remote_size': 0, 'remote_size_human': '0 B',
'warning': _STORAGE_UNAVAILABLE}
return jsonify(stats), 200
+6 -2
View File
@@ -12,6 +12,7 @@ Two endpoints:
* POST /generate — pure: given a plan dict, return Dockerfile + compose.
"""
import logging
import os
import shutil
import tempfile
@@ -25,6 +26,8 @@ from app.services.source_connection_service import SourceConnectionService
from app.middleware.rbac import developer_required, get_current_user
from app.api._request import json_object
logger = logging.getLogger(__name__)
buildpacks_bp = Blueprint('buildpacks', __name__)
@@ -113,8 +116,9 @@ def detect():
plan = BuildpackService.detect(clone_target)
BuildpackService.cache_plan(cache_key_url, plan, commit or branch)
return jsonify(_build_response(plan, app_name)), 200
except Exception as exc: # pragma: no cover - defensive
return jsonify({'error': str(exc)}), 400
except Exception: # pragma: no cover - defensive
logger.exception('Buildpack detection failed')
return jsonify({'error': 'Could not detect a build plan for this repository'}), 400
finally:
shutil.rmtree(workdir, ignore_errors=True)
+25 -14
View File
@@ -1,3 +1,5 @@
import logging
from flask import Blueprint, request, jsonify
from flask_jwt_extended import jwt_required, get_jwt_identity
from app.models import User, Application
@@ -7,6 +9,8 @@ from app import db, paths
from app.error_reporting import unexpected_response
from app.api._request import json_object
logger = logging.getLogger(__name__)
docker_bp = Blueprint('docker', __name__)
@@ -505,8 +509,9 @@ def docker_cleanup():
}
else:
results['containers'] = {'removed': 0, 'success': True}
except Exception as e:
results['containers'] = {'error': str(e)}
except Exception:
logger.exception('Docker cleanup step failed')
results['containers'] = {'error': 'Cleanup step failed; see server log'}
# Remove dangling images
try:
@@ -515,8 +520,9 @@ def docker_cleanup():
'success': result['success'],
'output': result['output'].strip()
}
except Exception as e:
results['images'] = {'error': str(e)}
except Exception:
logger.exception('Docker cleanup step failed')
results['images'] = {'error': 'Cleanup step failed; see server log'}
# Remove unused networks
try:
@@ -525,8 +531,9 @@ def docker_cleanup():
'success': result['success'],
'output': result['output'].strip()
}
except Exception as e:
results['networks'] = {'error': str(e)}
except Exception:
logger.exception('Docker cleanup step failed')
results['networks'] = {'error': 'Cleanup step failed; see server log'}
# Remove unused volumes (optional)
if include_volumes:
@@ -536,8 +543,9 @@ def docker_cleanup():
'success': result['success'],
'output': result['output'].strip()
}
except Exception as e:
results['volumes'] = {'error': str(e)}
except Exception:
logger.exception('Docker cleanup step failed')
results['volumes'] = {'error': 'Cleanup step failed; see server log'}
else:
results['volumes'] = {'skipped': True, 'message': 'Volume cleanup not requested'}
@@ -581,24 +589,27 @@ def cleanup_all_apps():
try:
compose_result = DockerService.compose_down(app.root_path, volumes=True, remove_orphans=True)
app_result['steps'].append({'action': 'docker_down', 'success': compose_result.get('success', False)})
except Exception as e:
app_result['steps'].append({'action': 'docker_down', 'error': str(e)})
except Exception:
logger.exception('App cleanup step failed')
app_result['steps'].append({'action': 'docker_down', 'error': 'Step failed; see server log'})
app_result['success'] = False
# Delete folder
try:
shutil.rmtree(app.root_path)
app_result['steps'].append({'action': 'delete_folder', 'success': True})
except Exception as e:
app_result['steps'].append({'action': 'delete_folder', 'error': str(e)})
except Exception:
logger.exception('App cleanup step failed')
app_result['steps'].append({'action': 'delete_folder', 'error': 'Step failed; see server log'})
app_result['success'] = False
# Remove from database
try:
db.session.delete(app)
app_result['steps'].append({'action': 'delete_db', 'success': True})
except Exception as e:
app_result['steps'].append({'action': 'delete_db', 'error': str(e)})
except Exception:
logger.exception('App cleanup step failed')
app_result['steps'].append({'action': 'delete_db', 'error': 'Step failed; see server log'})
app_result['success'] = False
results.append(app_result)
+4 -4
View File
@@ -883,15 +883,15 @@ def get_server_metrics(server_id):
try:
from_dt = datetime.fromisoformat(from_time.replace('Z', '+00:00'))
query = query.filter(ServerMetrics.timestamp >= from_dt)
except:
pass
except ValueError:
raise ValidationError("Invalid 'from' timestamp; use ISO 8601")
if to_time:
try:
to_dt = datetime.fromisoformat(to_time.replace('Z', '+00:00'))
query = query.filter(ServerMetrics.timestamp <= to_dt)
except:
pass
except ValueError:
raise ValidationError("Invalid 'to' timestamp; use ISO 8601")
metrics = query.order_by(ServerMetrics.timestamp.desc()).limit(limit).all()
+2 -1
View File
@@ -409,7 +409,8 @@ def _compare_versions(v1, v2):
elif a > b:
return 1
return 0
except:
except (ValueError, AttributeError, TypeError):
current_app.logger.warning('Could not compare versions %r and %r', v1, v2)
return 0
+9 -4
View File
@@ -1,4 +1,5 @@
"""SSO / OAuth API blueprint."""
import logging
from datetime import datetime, timedelta
from flask import Blueprint, request, jsonify, session
from flask_jwt_extended import (
@@ -16,6 +17,8 @@ from app.error_reporting import record_unexpected
from app.exceptions import NotFoundError
from app.api._request import json_object
logger = logging.getLogger(__name__)
sso_bp = Blueprint('sso', __name__)
VALID_PROVIDERS = ('google', 'github', 'oidc', 'saml')
@@ -47,8 +50,9 @@ def authorize(provider):
try:
auth_url, state = sso_service.generate_auth_url(provider, redirect_uri)
return jsonify({'auth_url': auth_url, 'state': state}), 200
except Exception as e:
return jsonify({'error': str(e)}), 400
except Exception:
logger.exception('SSO authorize failed for %s', provider)
return jsonify({'error': 'Could not start single sign-on'}), 400
@sso_bp.route('/callback/<provider>', methods=['POST'])
@@ -169,8 +173,9 @@ def link_provider(provider):
try:
profile = sso_service.handle_oauth_callback(provider, code, state, redirect_uri)
except Exception as e:
return jsonify({'error': str(e)}), 400
except Exception:
logger.exception('SSO link callback failed for %s', provider)
return jsonify({'error': 'Single sign-on failed'}), 400
# Check if this identity is already linked to another user
existing = OAuthIdentity.query.filter_by(
+3 -2
View File
@@ -84,7 +84,7 @@ def check_update():
def parse_version(v):
try:
return tuple(int(x) for x in v.split('.'))
except:
except ValueError:
return (0, 0, 0)
if parse_version(latest_version) > parse_version(current_version):
@@ -96,7 +96,8 @@ def check_update():
else:
result['error'] = f'GitHub API error: {e.code}'
except Exception as e:
result['error'] = f'Failed to check for updates: {str(e)}'
current_app.logger.warning('Update check failed: %s', e)
result['error'] = 'Failed to check for updates'
# Cache the result
_update_cache['last_check'] = time.time()
+6 -2
View File
@@ -6,6 +6,7 @@ Routes are mounted at ``/api/v1/waf``. Reads require access to the target app;
mutations and install require an admin user (mirrors ``app/api/dns_zones.py``).
Service ``ValueError``s map to HTTP 400.
"""
import logging
from flask import Blueprint, request, jsonify
from flask_jwt_extended import jwt_required
@@ -14,6 +15,8 @@ from app.middleware.rbac import require_admin_user
from app.api._access import require_app
from app.api._request import json_object
logger = logging.getLogger(__name__)
waf_bp = Blueprint('waf', __name__)
@@ -47,8 +50,9 @@ def update_policy(app_id):
# ModSecurity isn't present on the host.
try:
apply_result = WafService.apply(app_id)
except Exception as e: # pragma: no cover - defensive
apply_result = {'success': False, 'error': str(e)}
except Exception: # pragma: no cover - defensive
logger.exception('WAF apply failed for app %s', app_id)
apply_result = {'success': False, 'error': 'Failed to apply WAF policy'}
response = policy.to_dict()
response['apply'] = apply_result
+9 -1
View File
@@ -2,6 +2,10 @@
from flask import Flask
# sha256 of the build-injected import map (frontend/vite.config.js, from
# src/plugins/runtime/vendorManifest.js). test_security_headers recomputes it.
VENDOR_IMPORTMAP_HASH = 'sha256-lIJXjhqB4GwFLcjRTSjM91Q7vwXxH4Wbcn475CzkW4U='
def register_security_headers(app: Flask):
"""Register security headers as an after_request handler."""
@@ -37,9 +41,13 @@ def register_security_headers(app: Flask):
"form-action 'self' https://github.com",
]
else:
# style-src keeps 'unsafe-inline' because React renders style="" attributes;
# script-src has no 'unsafe-inline'/'unsafe-eval': the FOUC snippet is
# /theme-init.js; the one inline script (build-injected vendor importmap)
# is allowed by hash; blob: is for runtime extensions (plugins/runtime/loader.js).
csp_directives = [
"default-src 'self'",
"script-src 'self' 'unsafe-inline' 'unsafe-eval'",
"script-src 'self' '%s' blob:" % VENDOR_IMPORTMAP_HASH,
"style-src 'self' 'unsafe-inline'",
"img-src 'self' data: https:",
"font-src 'self'",
+6 -4
View File
@@ -19,10 +19,9 @@ import logging
import re
from datetime import datetime
import requests
from app import db
from app.models.chat_webhook import ChatWebhookConnection
from app.utils import safe_http
from app.utils.crypto import encrypt_secret
logger = logging.getLogger(__name__)
@@ -405,8 +404,11 @@ class ChatWebhookService:
headers[SIGNATURE_HEADER] = 'sha256=' + signature
try:
resp = requests.post(url, data=body, headers=headers,
timeout=_REQUEST_TIMEOUT)
resp = safe_http.post(url, data=body, headers=headers,
timeout=_REQUEST_TIMEOUT)
except safe_http.UnsafeURLError as exc:
# Non-retryable: the URL itself is the problem, retrying can't help.
return DeliveryResult.skipped(f'webhook url refused: {exc}')
except Exception as exc: # network error -> retryable failure
return DeliveryResult.failed(str(exc))
+4 -2
View File
@@ -214,7 +214,7 @@ class MonitoringService:
@classmethod
def send_webhook_alert(cls, alerts: List[Dict]) -> Dict:
"""Send webhook alert notification."""
import requests
from app.utils import safe_http
config = cls.get_config()
webhook_config = config.get('webhook', {})
@@ -229,7 +229,7 @@ class MonitoringService:
'alerts': alerts
}
response = requests.post(
response = safe_http.post(
webhook_config['url'],
json=payload,
timeout=10
@@ -239,6 +239,8 @@ class MonitoringService:
return {'success': True, 'message': 'Webhook sent'}
return {'success': False, 'error': f'Webhook returned {response.status_code}'}
except safe_http.UnsafeURLError as e:
return {'success': False, 'error': f'Webhook URL refused: {e}'}
except Exception as e:
return {'success': False, 'error': str(e)}
+2 -2
View File
@@ -214,13 +214,13 @@ SSL_BLOCK = '''
add_header Strict-Transport-Security "max-age=63072000; includeSubDomains; preload" always;
# Baseline response-header hardening. These are safe for reverse-proxied apps;
# the CSP is deliberately permissive so it does not break managed apps while
# the CSP is permissive (inline allowed, no eval) so it does not break managed apps while
# still satisfying "a CSP is present" and blocking cross-origin framing.
# Tighten the CSP per-app when the upstream can support a stricter policy.
add_header X-Content-Type-Options "nosniff" always;
add_header X-Frame-Options "SAMEORIGIN" always;
add_header Referrer-Policy "strict-origin-when-cross-origin" always;
add_header Content-Security-Policy "default-src 'self' 'unsafe-inline' 'unsafe-eval' data: blob: https:; frame-ancestors 'self'; upgrade-insecure-requests" always;
add_header Content-Security-Policy "default-src 'self' 'unsafe-inline' data: blob: https:; frame-ancestors 'self'; upgrade-insecure-requests" always;
'''
SSL_REDIRECT_TEMPLATE = '''server {{
+10 -6
View File
@@ -5,6 +5,8 @@ import threading
import time
import queue
import re
import logging
import subprocess
from app.services.system_service import SystemService
from app.services.log_service import LogService, LogStreamer
@@ -12,6 +14,8 @@ from app.services.docker_service import DockerService
from app import sockets_rooms as rooms
from app.utils.background_loop import BackgroundLoop
logger = logging.getLogger(__name__)
class AuthorizedSocketIO(SocketIO):
"""Recheck audience authorization before every server-side delivery.
@@ -775,11 +779,11 @@ def handle_subscribe_container_logs(data):
try:
process.terminate()
process.wait(timeout=2)
except:
except (OSError, subprocess.SubprocessError):
try:
process.kill()
except:
pass
except OSError as kill_err:
logger.debug('Could not kill log stream process: %s', kill_err)
thread = threading.Thread(target=stream_logs, daemon=True)
thread.start()
@@ -836,11 +840,11 @@ def stop_container_log_stream(sid: str):
try:
process.terminate()
process.wait(timeout=2)
except:
except (OSError, subprocess.SubprocessError):
try:
process.kill()
except:
pass
except OSError as kill_err:
logger.debug('Could not kill log stream process: %s', kill_err)
def emit_container_log(app_id: int, line: str, level: str = 'info'):
+1 -1
View File
@@ -1 +1 @@
5279
6102
+8 -8
View File
@@ -1,10 +1,10 @@
{
"php": "e6d77fd5ac90d438b9f694b3dcf50d63746bd87f3e20030e0c4bd6e94c6f372a",
"wordpress": "e6d77fd5ac90d438b9f694b3dcf50d63746bd87f3e20030e0c4bd6e94c6f372a",
"flask": "dba3ba47c2d2407bcbcf3b8014bc6f5dc105db8a7e5837f849b871fd5960329b",
"django": "dba3ba47c2d2407bcbcf3b8014bc6f5dc105db8a7e5837f849b871fd5960329b",
"python": "dba3ba47c2d2407bcbcf3b8014bc6f5dc105db8a7e5837f849b871fd5960329b",
"docker": "a18576b49626ed0f88126de8946163720d715016ee24119de22afdb38fda4400",
"remote": "7bc871f2c95a333cfe14129ae40df409a3a2165e79fb7c9205510957c95af99f",
"static": "91d63c4279f9fa1fa1ecc1cfd0e3f8c5b63e134613d319ae189a4560d3738e28"
"php": "a1de4fd08a9d277ecf4c7aaa46514bcae28a05b0a90a87bc2190cfa18dbb237f",
"wordpress": "a1de4fd08a9d277ecf4c7aaa46514bcae28a05b0a90a87bc2190cfa18dbb237f",
"flask": "15ffdb7f2221e465689f1662d49cf913861032d1ba1689f71ed78cd0dc2d963c",
"django": "15ffdb7f2221e465689f1662d49cf913861032d1ba1689f71ed78cd0dc2d963c",
"python": "15ffdb7f2221e465689f1662d49cf913861032d1ba1689f71ed78cd0dc2d963c",
"docker": "362d30566ba5452cb3ed9e7376da4f95c1fda61e6ec5b6a65b1ef98aa461aca5",
"remote": "0589dd632da88d569e41c339681c0a38656037d40b733179986bfc5f80f5bd27",
"static": "d3ff984ff1aac1c22e8e59f70f7fdf0c3d838726b4621f3313aa7cd4d78ee139"
}
@@ -0,0 +1,33 @@
"""GET /backups must not 500 when the backup directory cannot be created."""
import pytest
from app.services.backup_service import BackupService
@pytest.fixture
def unwritable(monkeypatch):
def deny():
raise PermissionError(13, 'Permission denied', '/var/backups/serverkit')
monkeypatch.setattr(BackupService, 'ensure_backup_dirs', deny)
@pytest.mark.parametrize('path', ['/api/v1/backups', '/api/v1/backups/stats'])
def test_get_degrades_with_a_warning(client, auth_headers, unwritable, path):
resp = client.get(path, headers=auth_headers)
assert resp.status_code == 200
body = resp.get_json()
assert 'not accessible' in body['warning']
assert '/var/backups' not in body['warning']
assert body.get('backups', []) == []
def test_create_returns_503_with_code_when_dir_unwritable(client, auth_headers, unwritable):
resp = client.post('/api/v1/backups/database', headers=auth_headers,
json={'db_type': 'mysql', 'db_name': 'x'})
assert resp.status_code == 503
body = resp.get_json()
assert body['code'] == 'backup_dir_unavailable'
assert body['error'] == 'Backup directory is not writable'
assert '/var/backups' not in resp.get_data(as_text=True)
@@ -0,0 +1,13 @@
"""Unexpected failures answer with a generic message; the detail is only logged."""
from app.services import sso_service
def test_sso_authorize_does_not_echo_exception_text(client, monkeypatch):
def boom(*a, **kw):
raise RuntimeError('secret internal path /etc/serverkit/oidc.key')
monkeypatch.setattr(sso_service, 'generate_auth_url', boom)
resp = client.get('/api/v1/sso/authorize/google?redirect_uri=http://localhost/cb')
assert resp.status_code == 400
assert '/etc/serverkit' not in resp.get_data(as_text=True)
@@ -321,7 +321,7 @@ class TestSigningAndDelivery:
captured.update({'url': url, 'data': data, 'headers': headers})
return _Resp()
monkeypatch.setattr('app.services.chat_webhook_service.requests.post', fake_post)
monkeypatch.setattr('app.services.chat_webhook_service.safe_http.post', fake_post)
user = _make_user()
conn = ChatWebhookService.add({'kind': 'webhook', 'name': 'Ops', 'url': 'https://hooks/x',
@@ -392,7 +392,7 @@ class TestConnectionTesting:
status_code = 503
monkeypatch.setattr(
'app.services.chat_webhook_service.requests.post',
'app.services.chat_webhook_service.safe_http.post',
lambda *args, **kwargs: _Resp(),
)
conn = ChatWebhookService.add({
@@ -428,7 +428,7 @@ class TestConnectionTesting:
raise RuntimeError(f'failed POST {url}')
monkeypatch.setattr(
'app.services.chat_webhook_service.requests.post', fail_post,
'app.services.chat_webhook_service.safe_http.post', fail_post,
)
secret_url = 'https://hooks.example/super-secret-token'
conn = ChatWebhookService.add({
@@ -454,7 +454,7 @@ class TestConnectionTesting:
raise RuntimeError(f'failed POST {url}')
monkeypatch.setattr(
'app.services.chat_webhook_service.requests.post', fail_post,
'app.services.chat_webhook_service.safe_http.post', fail_post,
)
conn = ChatWebhookService.add({
'kind': 'webhook',
@@ -624,7 +624,7 @@ class TestApi:
status_code = 503
monkeypatch.setattr(
'app.services.chat_webhook_service.requests.post',
'app.services.chat_webhook_service.safe_http.post',
lambda *args, **kwargs: _Resp(),
)
conn = ChatWebhookService.add({
+107
View File
@@ -0,0 +1,107 @@
"""PermissionService + User permission resolution per role."""
import pytest
from werkzeug.security import generate_password_hash
from app import db
from app.models import User
from app.services.permission_service import PermissionService
def _user(role, name=None):
u = User(email=f'{name or role}@example.com', username=name or role,
password_hash=generate_password_hash('x'), role=role, is_active=True)
db.session.add(u)
db.session.commit()
return u
def test_role_templates_cover_every_feature_for_admin():
admin = PermissionService.get_role_template('admin')
assert set(admin) == set(User.PERMISSION_FEATURES)
assert all(v == {'read': True, 'write': True} for v in admin.values())
def test_unknown_role_has_no_permissions(app):
assert PermissionService.get_role_template('wizard') == {}
u = _user('viewer', 'nobody')
u.role = 'wizard'
assert not u.has_permission('files')
def test_developer_defaults_follow_template(app):
dev = _user('developer')
assert dev.has_permission('applications', 'write')
assert dev.has_permission('monitoring', 'read')
assert not dev.has_permission('monitoring', 'write')
assert not dev.has_permission('users') # not in the developer template
def test_admin_always_allowed(app):
admin = _user('admin')
assert all(admin.has_permission(f, 'write') for f in User.PERMISSION_FEATURES)
assert admin.has_permission('anything-at-all')
def test_viewer_cannot_write_anything(app):
viewer = _user('viewer')
assert not any(viewer.has_permission(f, 'write') for f in User.PERMISSION_FEATURES)
def test_get_user_permissions_missing_user(app):
assert PermissionService.get_user_permissions(999999) is None
def test_update_custom_permissions_overrides_role(app):
dev = _user('developer')
res = PermissionService.update_user_permissions(
dev.id, {'files': {'read': True, 'write': False}})
assert res['success']
assert res['permissions'] == {'files': {'read': True, 'write': False}}
assert not dev.has_permission('files', 'write')
assert not dev.has_permission('applications') # custom set replaces the template wholesale
def test_reset_restores_role_template(app):
dev = _user('developer')
PermissionService.update_user_permissions(dev.id, {'files': {'read': True}})
res = PermissionService.reset_to_role_defaults(dev.id)
assert res['success'] and res['permissions'] == User.ROLE_PERMISSION_TEMPLATES['developer']
assert dev.permissions is None
def test_admin_permissions_cannot_be_customized(app):
admin = _user('admin')
res = PermissionService.update_user_permissions(admin.id, {'files': {'read': False}})
assert res == {'success': False, 'error': 'Admin permissions cannot be customized'}
def test_unknown_user_is_reported(app):
assert PermissionService.update_user_permissions(424242, {}) == {
'success': False, 'error': 'User not found'}
assert PermissionService.reset_to_role_defaults(424242) == {
'success': False, 'error': 'User not found'}
@pytest.mark.parametrize('perms, fragment', [
('nope', 'must be an object'),
({'bogus': {'read': True}}, 'Unknown feature'),
({'files': True}, 'read/write object'),
({'files': {'execute': True}}, 'Invalid permission key'),
({'files': {'read': 'yes'}}, 'must be boolean'),
({'files': {'read': False, 'write': True}}, 'without read'),
])
def test_validate_permissions_rejects(perms, fragment):
assert fragment in PermissionService.validate_permissions(perms)
def test_validate_permissions_accepts_good_input():
assert PermissionService.validate_permissions({}) is None
assert PermissionService.validate_permissions(
{'files': {'read': True, 'write': True}, 'cron': {'read': True}}) is None
def test_invalid_update_does_not_persist(app):
dev = _user('developer')
res = PermissionService.update_user_permissions(dev.id, {'bogus': {}})
assert not res['success']
assert dev.permissions is None
+48
View File
@@ -20,6 +20,14 @@ PANEL_NGINX_CONFIGS = [
/ 'sites-available'
/ 'serverkit.conf',
]
ROOT = Path(__file__).resolve().parents[2]
ALL_CSP_CONFIGS = PANEL_NGINX_CONFIGS + [
ROOT / 'nginx' / 'sites-available' / 'example.conf.template',
]
def _script_src(csp):
return next(d.strip() for d in csp.split(';') if d.strip().startswith('script-src '))
def _make_app(hsts_enabled, debug):
@@ -86,3 +94,43 @@ def test_resolve_ssl_mode(monkeypatch):
assert cfg._resolve_ssl_mode() == 'insecure'
monkeypatch.delenv('SERVERKIT_SSL_MODE', raising=False)
assert cfg._resolve_ssl_mode() == 'insecure'
def test_prod_csp_script_src_is_strict():
r = _make_app(hsts_enabled=False, debug=False).test_client().get('/ping')
script_src = _script_src(r.headers['Content-Security-Policy'])
assert 'unsafe-eval' not in script_src
assert 'unsafe-inline' not in script_src
@pytest.mark.parametrize('config_path', ALL_CSP_CONFIGS)
def test_nginx_csp_script_src_is_strict(config_path):
line = next(
l for l in config_path.read_text(encoding='utf-8').splitlines()
if 'Content-Security-Policy' in l
)
script_src = _script_src(line.split('"')[1])
assert 'unsafe-eval' not in script_src
assert 'unsafe-inline' not in script_src
def test_index_html_has_no_inline_script():
html = (ROOT / 'frontend' / 'index.html').read_text(encoding='utf-8')
assert '<script>' not in html
def test_csp_importmap_hash_matches_vendor_manifest():
import base64, hashlib, json, re
from app.middleware.security import VENDOR_IMPORTMAP_HASH
src = (ROOT / 'frontend/src/plugins/runtime/vendorManifest.js').read_text(encoding='utf-8')
pairs = re.findall(r"specifier: '([^']+)', file: '([^']+)'", src)
body = json.dumps({'imports': {s: '/serverkit-vendor/' + f for s, f in pairs}}, separators=(',', ':'))
digest = base64.b64encode(hashlib.sha256(body.encode()).digest()).decode()
assert VENDOR_IMPORTMAP_HASH == 'sha256-' + digest
for path in ALL_CSP_CONFIGS:
assert VENDOR_IMPORTMAP_HASH in path.read_text(encoding='utf-8')
def test_managed_app_nginx_template_has_no_unsafe_eval():
from app.services.nginx import templates
assert 'unsafe-eval' not in Path(templates.__file__).read_text(encoding='utf-8')
@@ -108,3 +108,17 @@ def test_list_servers_metrics_are_batched_not_per_row(client, auth_headers, serv
assert count_metric_queries() == baseline, (
'metrics lookup grew with the server count — it must stay batched'
)
@pytest.mark.parametrize('param', ['from', 'to'])
def test_metrics_rejects_a_malformed_timestamp_with_400(app, client, auth_headers, param):
with app.app_context():
server = _mk_server('delta')
db.session.commit()
server_id = server.id
resp = client.get(f'/api/v1/servers/{server_id}/metrics?{param}=not-a-date',
headers=auth_headers)
assert resp.status_code == 400
assert resp.get_json()['error']
+174
View File
@@ -0,0 +1,174 @@
"""TOTPService / TwoFactorSetup behaviour (RFC 6238 via pyotp)."""
import time
from urllib.parse import parse_qs, urlparse
import pyotp
import pytest
from werkzeug.security import generate_password_hash
from app import db
from app.models import User
from app.services.totp_service import TOTPService, TwoFactorSetup
@pytest.fixture
def secret():
return TOTPService.generate_secret()
@pytest.fixture
def user(app):
u = User(email='t@example.com', username='totp-user',
password_hash=generate_password_hash('x'), role='developer', is_active=True)
db.session.add(u)
db.session.commit()
return u
def test_generate_secret_is_random_valid_base32():
a, b = TOTPService.generate_secret(), TOTPService.generate_secret()
assert a != b
assert len(a) == 32
pyotp.TOTP(a).now() # decodes as base32
def test_current_code_verifies(secret):
code = TOTPService.get_current_code(secret)
assert len(code) == 6 and code.isdigit()
assert TOTPService.verify_code(secret, code)
def test_code_with_spaces_and_dashes_is_normalised(secret):
code = TOTPService.get_current_code(secret)
assert TOTPService.verify_code(secret, f'{code[:3]} {code[3:]}')
assert TOTPService.verify_code(secret, f'{code[:3]}-{code[3:]}')
@pytest.mark.parametrize('bad', ['', None, 'abcdef', '12345', '1234567', '000000x'])
def test_malformed_or_missing_code_is_rejected(secret, bad):
# A wrong 6-digit code could collide with the real one 1-in-a-million; use a shape check only.
assert not TOTPService.verify_code(secret, bad)
def test_missing_secret_is_rejected():
assert not TOTPService.verify_code(None, '123456')
assert not TOTPService.verify_code('', '123456')
def test_wrong_code_is_rejected(secret):
real = TOTPService.get_current_code(secret)
wrong = f'{(int(real) + 1) % 1000000:06d}'
# Neighbouring-window codes are also valid, so make sure we truly picked a stranger.
totp = TOTPService.get_totp(secret)
now = time.time()
valid = {totp.at(now + d * 30) for d in (-1, 0, 1)}
if wrong in valid:
pytest.skip('collided with a neighbouring window')
assert not TOTPService.verify_code(secret, wrong)
def test_window_drift(secret):
totp = TOTPService.get_totp(secret)
prev = totp.at(time.time() - 30)
far = totp.at(time.time() - 120)
assert TOTPService.verify_code(secret, prev, valid_window=1)
assert not TOTPService.verify_code(secret, far, valid_window=1)
assert not TOTPService.verify_code(secret, far, valid_window=0)
def test_provisioning_uri(secret):
uri = TOTPService.get_provisioning_uri(secret, 'me@example.com')
parsed = urlparse(uri)
assert parsed.scheme == 'otpauth' and parsed.netloc == 'totp'
qs = parse_qs(parsed.query)
assert qs['secret'] == [secret]
assert qs['issuer'] == ['ServerKit']
assert 'me%40example.com' in parsed.path or 'me@example.com' in parsed.path
def test_qr_code_is_png_data_uri(secret):
qr = TOTPService.generate_qr_code_base64(secret, 'me@example.com')
if qr is None:
pytest.skip('qrcode not installed')
assert qr.startswith('data:image/png;base64,')
def test_backup_codes_roundtrip():
plain, hashed = TOTPService.generate_backup_codes()
assert len(plain) == len(hashed) == TOTPService.BACKUP_CODE_COUNT
assert len(set(plain)) == len(plain)
assert all(len(c) == TOTPService.BACKUP_CODE_LENGTH for c in plain)
assert TOTPService.verify_backup_code(plain[0], hashed) == hashed[0]
assert TOTPService.verify_backup_code(plain[0].upper(), hashed) == hashed[0]
formatted = TOTPService.format_backup_codes([plain[0]])[0]
assert formatted == f'{plain[0][:4]}-{plain[0][4:]}'
assert TOTPService.verify_backup_code(formatted, hashed) == hashed[0]
def test_backup_code_rejects_unknown_and_empty():
_, hashed = TOTPService.generate_backup_codes()
assert TOTPService.verify_backup_code('deadbeef', hashed) is None
assert TOTPService.verify_backup_code('', hashed) is None
assert TOTPService.verify_backup_code('abcd1234', []) is None
def test_setup_flow_enables_2fa_and_returns_backup_codes(user):
setup = TwoFactorSetup(user)
assert setup.confirm_setup('123456') == (False, '2FA setup not initiated')
data = setup.initiate_setup()
assert data['issuer'] == 'ServerKit' and data['secret'] == user.totp_secret
assert not user.totp_enabled
ok, codes = setup.confirm_setup(TOTPService.get_current_code(user.totp_secret))
assert ok and len(codes) == 10
assert user.totp_enabled and user.totp_confirmed_at
assert setup.get_remaining_backup_codes_count() == 10
def test_confirm_with_wrong_code_does_not_enable(user):
setup = TwoFactorSetup(user)
setup.initiate_setup()
totp = TOTPService.get_totp(user.totp_secret)
now = time.time()
wrong = next(f'{n:06d}' for n in range(1000000)
if f'{n:06d}' not in {totp.at(now + d * 30) for d in (-1, 0, 1)})
assert setup.confirm_setup(wrong) == (False, 'Invalid verification code')
assert not user.totp_enabled
def _enabled(user):
setup = TwoFactorSetup(user)
setup.initiate_setup()
_, codes = setup.confirm_setup(TOTPService.get_current_code(user.totp_secret))
return setup, codes
def test_disable_with_totp_code(user):
setup, _ = _enabled(user)
ok, err = setup.disable(TOTPService.get_current_code(user.totp_secret))
assert ok and err is None
assert not user.totp_enabled and user.totp_secret is None and user.backup_codes is None
def test_disable_with_backup_code(user):
setup, codes = _enabled(user)
ok, err = setup.disable(codes[0])
assert ok and err is None and not user.totp_enabled
def test_disable_rejects_garbage_and_when_not_enabled(user):
setup = TwoFactorSetup(user)
assert setup.disable('123456') == (False, '2FA is not enabled')
setup, _ = _enabled(user)
assert setup.disable('zzzz-zzzz') == (False, 'Invalid verification code')
assert user.totp_enabled
def test_regenerate_backup_codes_replaces_old_set(user):
setup, old = _enabled(user)
old_hashes = set(user.get_backup_codes())
ok, new = setup.regenerate_backup_codes(TOTPService.get_current_code(user.totp_secret))
assert ok and len(new) == 10
assert not old_hashes & set(user.get_backup_codes())
assert setup.regenerate_backup_codes('abc') == (False, 'Invalid verification code')
+37
View File
@@ -0,0 +1,37 @@
"""User-configured webhook URLs must go through app.utils.safe_http."""
import pytest
from app.notifications.channels.base import DeliveryResult
from app.services.chat_webhook_service import ChatWebhookService
from app.services.monitoring_service import MonitoringService
PRIVATE_URLS = ['http://127.0.0.1:5000/hook', 'http://169.254.169.254/latest/meta-data', 'http://localhost/x']
@pytest.mark.parametrize('url', PRIVATE_URLS)
def test_monitoring_webhook_refuses_private_targets(url, monkeypatch):
monkeypatch.setattr(MonitoringService, 'get_config',
classmethod(lambda cls: {'webhook': {'enabled': True, 'url': url}}))
result = MonitoringService.send_webhook_alert([{'type': 'cpu'}])
assert result['success'] is False
assert 'refused' in result['error']
@pytest.mark.parametrize('url', PRIVATE_URLS)
def test_chat_webhook_refuses_private_targets(url):
class _Notification:
severity = 'info'
event_key = 'x'
title = 't'
body = 'b'
created_at = None
category = 'c'
action_path = None
id = 1
def get_data(self):
return {}
result = ChatWebhookService._deliver_webhook(None, {'url': url}, _Notification())
assert result.status == DeliveryResult.SKIPPED # terminal: the queue must not retry
assert 'refused' in result.error
+1 -27
View File
@@ -13,33 +13,7 @@
No Google Fonts / third-party CDN: a self-hosted control panel must not
leak every user's IP + usage to Google. Falls back to system-ui until
the woff2 files are present (run frontend/scripts/fetch-fonts.mjs). -->
<script>
// Prevent flash of unstyled content (FOUC) by setting theme + accent before React loads
(function() {
var d = document.documentElement;
d.setAttribute('data-theme', localStorage.getItem('theme') || 'dark');
var accent = localStorage.getItem('accent_color');
if (accent) {
var m = accent.match(/^#?([a-f\d]{2})([a-f\d]{2})([a-f\d]{2})$/i);
if (m) {
var r = parseInt(m[1],16), g = parseInt(m[2],16), b = parseInt(m[3],16);
var hx = function(v){ return Math.max(0,Math.min(255,v)).toString(16).padStart(2,'0'); };
var dk = function(v){ return Math.round(v*0.88); };
var br = function(v){ return Math.round(v + (255-v)*0.30); }; // toward white
var dm = function(v){ return Math.round(v*0.70); };
d.style.setProperty('--accent-primary', accent);
d.style.setProperty('--accent', accent);
d.style.setProperty('--accent-hover', '#'+hx(dk(r))+hx(dk(g))+hx(dk(b)));
d.style.setProperty('--accent-bright', '#'+hx(br(r))+hx(br(g))+hx(br(b)));
d.style.setProperty('--accent-dim', '#'+hx(dm(r))+hx(dm(g))+hx(dm(b)));
d.style.setProperty('--accent-bg', 'rgba('+r+','+g+','+b+',0.13)');
d.style.setProperty('--accent-bg-soft', 'rgba('+r+','+g+','+b+',0.07)');
d.style.setProperty('--accent-glow', 'rgba('+r+','+g+','+b+',0.35)');
d.style.setProperty('--accent-shadow', 'rgba('+r+','+g+','+b+',0.3)');
}
}
})();
</script>
<script src="/theme-init.js"></script>
</head>
<body>
+26
View File
@@ -0,0 +1,26 @@
// Prevent flash of unstyled content (FOUC) by setting theme + accent before React loads.
// External file (not inline) so the CSP needs no 'unsafe-inline' for scripts.
(function() {
var d = document.documentElement;
d.setAttribute('data-theme', localStorage.getItem('theme') || 'dark');
var accent = localStorage.getItem('accent_color');
if (accent) {
var m = accent.match(/^#?([a-f\d]{2})([a-f\d]{2})([a-f\d]{2})$/i);
if (m) {
var r = parseInt(m[1],16), g = parseInt(m[2],16), b = parseInt(m[3],16);
var hx = function(v){ return Math.max(0,Math.min(255,v)).toString(16).padStart(2,'0'); };
var dk = function(v){ return Math.round(v*0.88); };
var br = function(v){ return Math.round(v + (255-v)*0.30); }; // toward white
var dm = function(v){ return Math.round(v*0.70); };
d.style.setProperty('--accent-primary', accent);
d.style.setProperty('--accent', accent);
d.style.setProperty('--accent-hover', '#'+hx(dk(r))+hx(dk(g))+hx(dk(b)));
d.style.setProperty('--accent-bright', '#'+hx(br(r))+hx(br(g))+hx(br(b)));
d.style.setProperty('--accent-dim', '#'+hx(dm(r))+hx(dm(g))+hx(dm(b)));
d.style.setProperty('--accent-bg', 'rgba('+r+','+g+','+b+',0.13)');
d.style.setProperty('--accent-bg-soft', 'rgba('+r+','+g+','+b+',0.07)');
d.style.setProperty('--accent-glow', 'rgba('+r+','+g+','+b+',0.35)');
d.style.setProperty('--accent-shadow', 'rgba('+r+','+g+','+b+',0.3)');
}
}
})();
+1
View File
@@ -217,6 +217,7 @@ const Backups = () => {
]);
setBackups(backupsRes.backups || []);
if (backupsRes.warning) setError(backupsRes.warning);
setStats(statsRes);
setConfig(configRes);
if (refreshSchedules) await reloadSchedules();
@@ -5,6 +5,7 @@ import { useEffect, useRef, useState, useCallback } from 'react';
import { useTranslation, useFormat } from 'serverkit-sdk';
import SyntheticDesktop from './SyntheticDesktop.jsx';
import { Button as SharedButton } from 'serverkit-sdk';
import { storageKey } from '../../../services/tunnelBase';
const FRAME_INTERVAL_MS_DEFAULT = 700;
const STORAGE_KEY = 'sk-gui:prefs';
@@ -52,7 +53,7 @@ export default function ServerGui({ api, serverId }) {
if (api && typeof api.request === 'function') {
return api.request(path);
}
const token = localStorage.getItem('access_token');
const token = localStorage.getItem(storageKey('access_token'));
const r = await fetch(path, { headers: token ? { Authorization: `Bearer ${token}` } : {} });
if (!r.ok) throw new Error(`HTTP ${r.status}`);
return r.json();
+1 -1
View File
@@ -111,7 +111,7 @@ http {
add_header Permissions-Policy "geolocation=(), microphone=(), camera=()" always;
# Content Security Policy
add_header Content-Security-Policy "default-src 'self'; script-src 'self' 'unsafe-inline' 'unsafe-eval'; style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self'; connect-src 'self' ws: wss:; frame-ancestors 'none'; base-uri 'self'; form-action 'self' https://github.com;" always;
add_header Content-Security-Policy "default-src 'self'; script-src 'self' 'sha256-lIJXjhqB4GwFLcjRTSjM91Q7vwXxH4Wbcn475CzkW4U=' blob:; style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self'; connect-src 'self' ws: wss:; frame-ancestors 'none'; base-uri 'self'; form-action 'self' https://github.com;" always;
# Connection limits
limit_conn conn_limit 20;
+1 -1
View File
@@ -58,7 +58,7 @@ server {
add_header X-XSS-Protection "1; mode=block" always;
add_header Referrer-Policy "strict-origin-when-cross-origin" always;
add_header Permissions-Policy "geolocation=(), microphone=(), camera=()" always;
add_header Content-Security-Policy "default-src 'self'; script-src 'self' 'unsafe-inline' 'unsafe-eval' blob:; style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self'; connect-src 'self' ws: wss:; frame-ancestors 'none'; base-uri 'self'; form-action 'self';" always;
add_header Content-Security-Policy "default-src 'self'; script-src 'self' 'sha256-lIJXjhqB4GwFLcjRTSjM91Q7vwXxH4Wbcn475CzkW4U=' blob:; style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self'; connect-src 'self' ws: wss:; frame-ancestors 'none'; base-uri 'self'; form-action 'self';" always;
# Option 1: Proxy to a local app (Docker container, Node.js, etc.)
location / {
+4 -3
View File
@@ -103,9 +103,10 @@ server {
add_header X-Robots-Tag "noindex, nofollow, noarchive" always;
# Content Security Policy
# Note: 'unsafe-inline'/'unsafe-eval' are kept because many SPAs need them.
# Once your build is verified to work without inline scripts, remove them.
add_header Content-Security-Policy "default-src 'self'; script-src 'self' 'unsafe-inline' 'unsafe-eval' blob:; style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self'; connect-src 'self' ws: wss:; frame-ancestors 'none'; base-uri 'self'; form-action 'self' https://github.com;" always;
# script-src has no 'unsafe-inline'/'unsafe-eval' (theme init is /theme-init.js,
# the vendor importmap is allowed by hash). style-src keeps 'unsafe-inline'
# because React renders style="" attributes.
add_header Content-Security-Policy "default-src 'self'; script-src 'self' 'sha256-lIJXjhqB4GwFLcjRTSjM91Q7vwXxH4Wbcn475CzkW4U=' blob:; style-src 'self' 'unsafe-inline'; img-src 'self' data: https:; font-src 'self'; connect-src 'self' ws: wss:; frame-ancestors 'none'; base-uri 'self'; form-action 'self' https://github.com;" always;
# ═══════════════════════════════════════════════════════════════════════════
# ServerKit frontend — served directly by host nginx from the built SPA