copenhagentruckwash/api
1
0
Fork 0
Code Issues Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

Remove edge_broker_client.php and related unit tests. Add new dependencies and configuration files for edge-agent, including build scripts and package-lock updates.

Browse Source
This commit is contained in:
Jeppe Bundgaard
2026-04-13 09:42:35 +02:00
parent 16c9cd9931
commit a83b850aba
340 changed files with 45323 additions and 880 deletions
Download Patch File Download Diff File Expand all files Collapse all files
services/nginx/app/classes/edge_gateway_manager.php
+508 -158
View File
@@ -10,6 +10,8 @@ use objects\edge_gateway_claim_tokens_o;
use objects\edge_gateway_command_jobs_o;
use objects\edge_gateway_device_inventory_o;
use objects\edge_gateway_relay_bindings_o;
use objects\edge_gateway_shell_action_jobs_o;
use objects\edge_gateway_shell_events_o;
use objects\edge_gateway_shell_sessions_o;
use objects\edge_gateway_update_jobs_o;
use objects\edge_gateways_o;
@@ -32,8 +34,12 @@ class edge_gateway_manager
public const COMMAND_POLL_TIMEOUT_SECONDS = 20;
public const COMMAND_POLL_INTERVAL_MICROSECONDS = 250000;
public const COMMAND_DISPATCH_STALE_AFTER_SECONDS = 30;
public const SHELL_ACTION_POLL_TIMEOUT_SECONDS = 20;
public const SHELL_ACTION_STALE_AFTER_SECONDS = 30;
public const SHELL_EVENT_POLL_TIMEOUT_SECONDS = 5;
public const SHELL_EVENT_POLL_LIMIT = 200;
public function __construct(private readonly ?edge_broker_client $brokerClient = null)
public function __construct()
{
edge_gateway_schema_bootstrap::ensureTables();
}
@@ -118,7 +124,6 @@ class edge_gateway_manager
return [
'gateway' => $this->getGateway($gatewayId),
'agent_token' => $agentToken,
'broker_url' => $this->getBrokerPublicUrl(),
'heartbeat_url' => $this->getApiBaseUrl() . '/edge-agent/gateways/' . $gatewayId . '/heartbeat',
'release_channel' => (string)$gateway->release_channel->value(),
];
@@ -443,8 +448,23 @@ class edge_gateway_manager
*/
public function createShellSession(int $gatewayId, string $reason, ?int $userId = null): array
{
$gateway = $this->requireGateway($gatewayId);
return $this->createShellSessionRequest($gatewayId, $reason, [], $userId);
}
/**
* @throws Exception
*/
public function createShellSessionRequest(int $gatewayId, string $reason, array $options = [], ?int $userId = null): array
{
$gateway = $this->requireDispatchableGateway($gatewayId);
$sessionToken = bin2hex(random_bytes(32));
$metadata = [
'ttl_seconds' => self::SHELL_SESSION_TTL_SECONDS,
'transport' => 'API_POLLING',
];
$cols = isset($options['cols']) ? (int)$options['cols'] : null;
$rows = isset($options['rows']) ? (int)$options['rows'] : null;
$sessionObject = new edge_gateway_shell_sessions_o();
$sessionId = $sessionObject->add_object([
@@ -456,11 +476,20 @@ class edge_gateway_manager
'approved_by' => $userId,
'approved_at' => $this->now(),
'expires_at' => $this->formatDateTime(time() + self::SHELL_SESSION_TTL_SECONDS),
'metadata_json' => [
'ttl_seconds' => self::SHELL_SESSION_TTL_SECONDS,
],
'metadata_json' => $metadata,
]);
$sessionObject->select($sessionId);
$this->createShellActionJob(
(int)$sessionObject->id,
$gatewayId,
'OPEN',
array_filter([
'reason' => $reason,
'cols' => $cols,
'rows' => $rows,
], static fn(mixed $value): bool => $value !== null && $value !== ''),
$userId
);
$this->writeAudit(
$gatewayId,
@@ -473,7 +502,7 @@ class edge_gateway_manager
return [
'session' => $sessionObject->asArray(),
'session_token' => $sessionToken,
'websocket_url' => $this->buildBrowserShellWsUrl($sessionToken),
'transport' => 'API_POLLING',
];
}
@@ -527,6 +556,225 @@ class edge_gateway_manager
return $session->asArray();
}
/**
* @throws Exception
*/
public function getShellSession(int $gatewayId, int $sessionId): array
{
return $this->requireShellSessionForGateway($gatewayId, $sessionId)->asArray();
}
/**
* @return array<string,mixed>
* @throws Exception
*/
public function pollShellSessionEvents(
int $gatewayId,
int $sessionId,
int $afterId = 0,
int $waitSeconds = self::SHELL_EVENT_POLL_TIMEOUT_SECONDS
): array {
$session = $this->requireShellSessionForGateway($gatewayId, $sessionId);
$deadline = microtime(true) + max(0, $waitSeconds);
$cursor = max(0, $afterId);
do {
$session = (new edge_gateway_shell_sessions_o())->select($sessionId);
$events = $this->listShellEvents($sessionId, $cursor, self::SHELL_EVENT_POLL_LIMIT);
if ($events !== [] || $session->closed_at->value() !== null) {
return [
'session' => $session->asArray(),
'events' => $events,
];
}
if (microtime(true) >= $deadline) {
break;
}
usleep(self::COMMAND_POLL_INTERVAL_MICROSECONDS);
} while (true);
return [
'session' => $session->asArray(),
'events' => [],
];
}
/**
* @throws Exception
*/
public function enqueueShellInput(int $gatewayId, int $sessionId, string $data, ?int $userId = null): array
{
$session = $this->requireActiveShellSessionForGateway($gatewayId, $sessionId);
$this->createShellActionJob(
$sessionId,
$gatewayId,
'INPUT',
['data' => $data],
$userId
);
return $session->asArray();
}
/**
* @throws Exception
*/
public function enqueueShellResize(int $gatewayId, int $sessionId, int $cols, int $rows, ?int $userId = null): array
{
$session = $this->requireActiveShellSessionForGateway($gatewayId, $sessionId);
$this->createShellActionJob(
$sessionId,
$gatewayId,
'RESIZE',
[
'cols' => max(1, $cols),
'rows' => max(1, $rows),
],
$userId
);
return $session->asArray();
}
/**
* @throws Exception
*/
public function requestShellSessionClose(int $gatewayId, int $sessionId, ?int $userId = null): array
{
$session = $this->requireShellSessionForGateway($gatewayId, $sessionId);
$metadata = (array)($session->metadata_json->value() ?? []);
$metadata['close_requested_by'] = $userId;
$session->metadata_json->set($metadata);
$wasClosed = $session->closed_at->value() !== null;
if (!$wasClosed) {
$session->approval_status->set('CLOSED');
$session->closed_at->set($this->now());
$this->writeAudit(
(int)$session->gateway_id->value(),
null,
'ROOT_SHELL_CLOSED',
$userId,
['session_id' => (int)$session->id, 'closed_reason' => 'browser_requested']
);
}
$this->cancelPendingShellActionJobs($sessionId);
$this->createShellActionJob(
$sessionId,
$gatewayId,
'CLOSE',
['reason' => 'browser_requested'],
$userId
);
$this->appendShellEvent(
(int)$session->gateway_id->value(),
(int)$session->id,
'CLOSED',
['reason' => 'browser_requested']
);
return $session->asArray();
}
/**
* @throws Exception
*/
public function pollShellAction(
int $gatewayId,
string $plainToken,
int $waitSeconds = self::SHELL_ACTION_POLL_TIMEOUT_SECONDS
): ?array {
$gateway = $this->authenticateGateway($gatewayId, $plainToken);
$deadline = microtime(true) + max(0, $waitSeconds);
do {
$job = $this->claimNextShellActionJob($gateway);
if ($job !== null) {
return $this->formatAgentShellAction($job);
}
if (microtime(true) >= $deadline) {
break;
}
usleep(self::COMMAND_POLL_INTERVAL_MICROSECONDS);
} while (true);
return null;
}
/**
* @throws Exception
*/
public function submitShellActionResult(
int $gatewayId,
int $actionId,
string $plainToken,
bool $ok,
?string $error = null
): array {
$gateway = $this->authenticateGateway($gatewayId, $plainToken);
$job = (new edge_gateway_shell_action_jobs_o())->select($actionId);
if (!$job->exists()) {
throw new Exception('Edge gateway shell action job not found');
}
if ((int)$job->gateway_id->value() !== (int)$gateway->id) {
throw new Exception('Edge gateway shell action job does not belong to this gateway');
}
$status = (string)$job->status->value();
if (in_array($status, ['COMPLETED', 'FAILED'], true)) {
return [
'acknowledged' => true,
'job' => $job->asArray(),
];
}
$job->status->set($ok ? 'COMPLETED' : 'FAILED');
$job->completed_at->set($this->now());
$job->error_message->set($ok ? null : (trim((string)$error) !== '' ? trim((string)$error) : 'Edge gateway shell action failed'));
return [
'acknowledged' => true,
'job' => $job->asArray(),
];
}
/**
* @param array<int,array<string,mixed>> $events
* @return array<string,mixed>
* @throws Exception
*/
public function recordShellSessionEvents(
int $gatewayId,
int $sessionId,
string $plainToken,
array $events
): array {
$gateway = $this->authenticateGateway($gatewayId, $plainToken);
$session = $this->requireShellSessionForGateway((int)$gateway->id, $sessionId);
$recordedEvents = [];
foreach ($events as $event) {
if (!is_array($event)) {
continue;
}
$recorded = $this->applyShellSessionEvent($session, $event);
if ($recorded !== null) {
$recordedEvents[] = $recorded;
}
}
return [
'session' => $session->asArray(),
'events' => $recordedEvents,
];
}
/**
* @throws Exception
*/
@@ -561,8 +809,6 @@ class edge_gateway_manager
'channel' => (int)$binding['channel'],
], null);
$this->tryImmediateBrokerDispatch($job, $gateway);
return $this->waitForCommandResult((int)$job->id);
}
@@ -581,8 +827,6 @@ class edge_gateway_manager
'on' => $on,
], null);
$this->tryImmediateBrokerDispatch($job, $gateway);
return $this->waitForCommandResult((int)$job->id);
}
@@ -627,11 +871,12 @@ class edge_gateway_manager
{
$configJson = json_encode([
'apiUrl' => $this->getApiBaseUrl(),
'brokerUrl' => $this->getBrokerPublicUrl(),
'installToken' => $plainToken,
'gatewayId' => null,
'agentToken' => null,
'heartbeatIntervalSeconds' => 15,
'commandPollTimeoutSeconds' => self::COMMAND_POLL_TIMEOUT_SECONDS,
'shellActionPollTimeoutSeconds' => self::SHELL_ACTION_POLL_TIMEOUT_SECONDS,
], JSON_UNESCAPED_SLASHES);
$script = <<<'BASH'
@@ -698,38 +943,6 @@ BASH;
];
}
public function buildBrowserShellWsUrl(string $sessionToken): ?string
{
$brokerUrl = $this->getBrokerPublicUrl();
if ($brokerUrl === null) {
return null;
}
$parsed = parse_url($brokerUrl);
if (!is_array($parsed) || !isset($parsed['host'])) {
return null;
}
$scheme = (($parsed['scheme'] ?? 'http') === 'https') ? 'wss' : 'ws';
$url = $scheme . '://' . $parsed['host'];
if (isset($parsed['port'])) {
$url .= ':' . $parsed['port'];
}
$url .= '/ws/browser-shell?token=' . urlencode($sessionToken);
return $url;
}
public function getBrokerPublicUrl(): ?string
{
$configured = trim((string)(getenv('EDGE_BROKER_PUBLIC_URL') ?: ''));
if ($configured !== '') {
return $this->normalizeBrokerPublicUrl($configured);
}
return $this->buildNormalizedBrokerOrigin($this->getApiBaseUrl());
}
public function getApiBaseUrl(): string
{
$configured = trim((string)(getenv('EDGE_PUBLIC_API_URL') ?: ''));
@@ -802,55 +1015,6 @@ BASH;
return null;
}
private function normalizeBrokerPublicUrl(string $url): string
{
$normalized = $this->buildNormalizedBrokerOrigin($url);
return $normalized ?? trim($url);
}
private function buildNormalizedBrokerOrigin(string $url): ?string
{
$parsed = parse_url(trim($url));
if (!is_array($parsed) || !isset($parsed['host'])) {
return null;
}
$host = (string)$parsed['host'];
$scheme = strtolower((string)($parsed['scheme'] ?? ''));
if ($scheme === '' || ($scheme === 'http' && $this->shouldUseSecureBrokerScheme($host))) {
$scheme = $this->shouldUseSecureBrokerScheme($host) ? 'https' : 'http';
}
$normalized = $scheme . '://' . $host;
if (isset($parsed['port'])) {
$normalized .= ':' . $parsed['port'];
}
return $normalized;
}
private function shouldUseSecureBrokerScheme(string $host): bool
{
$normalized = strtolower(trim($host, '[]'));
if ($normalized === '' || $normalized === 'localhost' || $normalized === 'edge-broker') {
return false;
}
if (str_ends_with($normalized, '.localhost')
|| str_ends_with($normalized, '.local')
|| str_ends_with($normalized, '.lan')
|| str_ends_with($normalized, '.internal')
|| str_ends_with($normalized, '.home.arpa')) {
return false;
}
if (filter_var($normalized, FILTER_VALIDATE_IP) !== false) {
return filter_var($normalized, FILTER_VALIDATE_IP, FILTER_FLAG_NO_PRIV_RANGE | FILTER_FLAG_NO_RES_RANGE) !== false;
}
return str_contains($normalized, '.');
}
/**
* @throws Exception
*/
@@ -982,38 +1146,262 @@ BASH;
return $jobObject->select($jobId);
}
private function createShellActionJob(
int $sessionId,
int $gatewayId,
string $actionType,
array $payload,
?int $userId
): edge_gateway_shell_action_jobs_o {
$jobObject = new edge_gateway_shell_action_jobs_o();
$jobId = $jobObject->add_object([
'gateway_id' => $gatewayId,
'session_id' => $sessionId,
'action_type' => strtoupper($actionType),
'status' => 'PENDING',
'payload_json' => $payload,
'requested_by' => $userId,
'requested_at' => $this->now(),
]);
return $jobObject->select($jobId);
}
private function claimNextShellActionJob(edge_gateways_o $gateway): ?edge_gateway_shell_action_jobs_o
{
$pdo = db::getPDO();
$pdo->beginTransaction();
try {
$statement = $pdo->prepare(
'SELECT id
FROM edge_gateway_shell_action_jobs
WHERE gateway_id = :gateway_id
AND deleted_at IS NULL
AND (
status = :pending_status_match
OR (
status = :dispatching_status_match
AND COALESCE(updated_at, created_at, requested_at) <= :stale_before
)
)
ORDER BY CASE WHEN status = :pending_status_order THEN 0 ELSE 1 END, requested_at ASC, id ASC
LIMIT 1
FOR UPDATE'
);
$statement->execute([
':gateway_id' => (int)$gateway->id,
':pending_status_match' => 'PENDING',
':dispatching_status_match' => 'DISPATCHING',
':pending_status_order' => 'PENDING',
':stale_before' => $this->formatDateTime(time() - self::SHELL_ACTION_STALE_AFTER_SECONDS),
]);
$row = $statement->fetch();
if (!is_array($row) || !isset($row['id'])) {
$pdo->commit();
return null;
}
$update = $pdo->prepare(
'UPDATE edge_gateway_shell_action_jobs
SET status = :status,
error_message = NULL,
completed_at = NULL
WHERE id = :id'
);
$update->execute([
':status' => 'DISPATCHING',
':id' => (int)$row['id'],
]);
$pdo->commit();
} catch (\Throwable $throwable) {
if ($pdo->inTransaction()) {
$pdo->rollBack();
}
throw $throwable;
}
return (new edge_gateway_shell_action_jobs_o())->select((int)$row['id']);
}
private function formatAgentShellAction(edge_gateway_shell_action_jobs_o $job): array
{
$session = $this->requireShellSessionForGateway((int)$job->gateway_id->value(), (int)$job->session_id->value());
return [
'id' => (int)$job->id,
'gateway_id' => (int)$job->gateway_id->value(),
'session_id' => (int)$job->session_id->value(),
'action_type' => (string)$job->action_type->value(),
'actionType' => (string)$job->action_type->value(),
'payload' => array_merge([
'sessionId' => (int)$session->id,
'reason' => (string)$session->reason->value(),
'expiresAt' => (string)$session->expires_at->value(),
], (array)($job->payload_json->value() ?? [])),
'requested_at' => (string)$job->requested_at->value(),
];
}
private function applyShellSessionEvent(edge_gateway_shell_sessions_o $session, array $event): ?array
{
$type = strtoupper(trim((string)($event['type'] ?? $event['event_type'] ?? '')));
if ($type === '') {
return null;
}
$payload = isset($event['payload']) && is_array($event['payload']) ? (array)$event['payload'] : [];
if (array_key_exists('data', $event) && !array_key_exists('data', $payload)) {
$payload['data'] = (string)$event['data'];
}
if (array_key_exists('code', $event) && !array_key_exists('code', $payload)) {
$payload['code'] = (int)$event['code'];
}
if (array_key_exists('reason', $event) && !array_key_exists('reason', $payload)) {
$payload['reason'] = (string)$event['reason'];
}
if ($type === 'OPENED') {
if ($session->opened_at->value() === null) {
$session->opened_at->set($this->now());
}
$session->approval_status->set('OPEN');
return $this->appendShellEvent((int)$session->gateway_id->value(), (int)$session->id, $type, $payload);
}
if ($type === 'OUTPUT') {
$data = (string)($payload['data'] ?? '');
if ($data === '') {
return null;
}
$session->transcript_text->set((string)($session->transcript_text->value() ?? '') . $data);
return $this->appendShellEvent(
(int)$session->gateway_id->value(),
(int)$session->id,
$type,
['data' => $data]
);
}
if ($type === 'CLOSED') {
$metadata = (array)($session->metadata_json->value() ?? []);
$closedReason = trim((string)($payload['reason'] ?? 'agent_closed'));
$metadata['closed_reason'] = $closedReason !== '' ? $closedReason : 'agent_closed';
if (array_key_exists('code', $payload)) {
$metadata['exit_code'] = (int)$payload['code'];
}
$session->metadata_json->set($metadata);
$wasClosed = $session->closed_at->value() !== null;
$session->approval_status->set('CLOSED');
if (!$wasClosed) {
$session->closed_at->set($this->now());
$this->writeAudit(
(int)$session->gateway_id->value(),
null,
'ROOT_SHELL_CLOSED',
null,
['session_id' => (int)$session->id, 'closed_reason' => $metadata['closed_reason']]
);
}
$this->cancelPendingShellActionJobs((int)$session->id);
return $this->appendShellEvent((int)$session->gateway_id->value(), (int)$session->id, $type, $payload);
}
return $this->appendShellEvent((int)$session->gateway_id->value(), (int)$session->id, $type, $payload);
}
private function appendShellEvent(int $gatewayId, int $sessionId, string $eventType, array $payload): array
{
$eventObject = new edge_gateway_shell_events_o();
$eventId = $eventObject->add_object([
'gateway_id' => $gatewayId,
'session_id' => $sessionId,
'event_type' => strtoupper($eventType),
'payload_json' => $payload,
]);
return $eventObject->select($eventId)->asArray();
}
/**
* @return array<int,array<string,mixed>>
*/
private function listShellEvents(int $sessionId, int $afterId = 0, int $limit = self::SHELL_EVENT_POLL_LIMIT): array
{
$safeAfterId = max(0, $afterId);
$safeLimit = max(1, min(self::SHELL_EVENT_POLL_LIMIT, $limit));
$statement = db::getPDO()->prepare(
'SELECT id
FROM edge_gateway_shell_events
WHERE session_id = :session_id
AND deleted_at IS NULL
AND id > :after_id
ORDER BY id ASC
LIMIT ' . $safeLimit
);
$statement->execute([
':session_id' => $sessionId,
':after_id' => $safeAfterId,
]);
$events = [];
while (($row = $statement->fetch()) && is_array($row) && isset($row['id'])) {
$events[] = (new edge_gateway_shell_events_o())->select((int)$row['id'])->asArray();
}
return $events;
}
/**
* @throws Exception
*/
private function tryImmediateBrokerDispatch(edge_gateway_command_jobs_o $job, edge_gateways_o $gateway): void
private function requireShellSessionForGateway(int $gatewayId, int $sessionId): edge_gateway_shell_sessions_o
{
$this->markCommandJobDispatching($job);
try {
$response = $this->broker()->dispatchCommand(
(int)$gateway->id,
(string)$job->command_type->value(),
$this->buildCommandExecutionPayload($job, $gateway)
);
$ok = (bool)($response['ok'] ?? false);
$payload = (array)($response['payload'] ?? []);
$errorMessage = $ok ? null : trim((string)($response['error'] ?? 'Edge broker command failed'));
$this->finalizeCommandJob($job, $ok, $payload, $errorMessage, $gateway);
if (!$ok) {
throw new Exception($errorMessage ?: 'Edge broker command failed');
}
} catch (\Throwable $throwable) {
if ($this->shouldFallbackToQueuedDelivery($throwable)) {
$this->releaseCommandJobToQueue($job);
return;
}
$this->finalizeCommandJob($job, false, [], $throwable->getMessage(), $gateway);
throw $throwable;
$session = (new edge_gateway_shell_sessions_o())->select($sessionId);
if (!$session->exists() || $session->deleted_at->value() !== null) {
throw new Exception('Edge gateway shell session not found');
}
if ((int)$session->gateway_id->value() !== $gatewayId) {
throw new Exception('Edge gateway shell session does not belong to this gateway');
}
return $session;
}
/**
* @throws Exception
*/
private function requireActiveShellSessionForGateway(int $gatewayId, int $sessionId): edge_gateway_shell_sessions_o
{
$session = $this->requireShellSessionForGateway($gatewayId, $sessionId);
if ($session->closed_at->value() !== null) {
throw new Exception('Edge gateway shell session is already closed');
}
if (strtotime((string)$session->expires_at->value()) < time()) {
throw new Exception('Edge gateway shell session has expired');
}
return $session;
}
private function cancelPendingShellActionJobs(int $sessionId): void
{
$statement = db::getPDO()->prepare(
'UPDATE edge_gateway_shell_action_jobs
SET deleted_at = :deleted_at
WHERE session_id = :session_id
AND deleted_at IS NULL
AND status IN (\'PENDING\', \'DISPATCHING\')'
);
$statement->execute([
':deleted_at' => $this->now(),
':session_id' => $sessionId,
]);
}
/**
@@ -1186,39 +1574,6 @@ BASH;
}
}
private function releaseCommandJobToQueue(edge_gateway_command_jobs_o $job): void
{
$job->response_json->set([]);
$job->error_message->set(null);
$job->completed_at->set(null);
$job->status->set('PENDING');
}
private function markCommandJobDispatching(edge_gateway_command_jobs_o $job): void
{
$job->response_json->set([]);
$job->error_message->set(null);
$job->completed_at->set(null);
$job->status->set('DISPATCHING');
}
private function shouldFallbackToQueuedDelivery(\Throwable $throwable): bool
{
if ($throwable instanceof edge_broker_transport_exception) {
return true;
}
if ($throwable instanceof edge_broker_http_exception) {
return $throwable->statusCode() === 503;
}
$message = $throwable->getMessage();
return str_contains($message, 'Gateway agent is offline')
|| str_contains($message, 'Could not resolve host:')
|| str_contains($message, 'Failed to connect')
|| str_contains($message, 'Connection refused');
}
private function markLinkedUpdateJobStarted(int $commandJobId): void
{
$updateJob = $this->findLinkedUpdateJobByCommandId($commandJobId);
@@ -1443,11 +1798,6 @@ BASH;
]);
}
private function broker(): edge_broker_client
{
return $this->brokerClient ?? new edge_broker_client();
}
private function hashToken(string $plainToken): string
{
return hash('sha256', $plainToken);