init
This commit is contained in:
@@ -0,0 +1,349 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace LayLink\Server;
|
||||
|
||||
use LayLink\Audit\AuditLogger;
|
||||
use LayLink\Auth\ClientAuthenticator;
|
||||
use LayLink\Auth\NodeAuthenticator;
|
||||
use LayLink\Node\NodeConnection;
|
||||
use LayLink\Node\NodeRegistry;
|
||||
use LayLink\Protocol\Frame;
|
||||
use LayLink\Protocol\FrameCodec;
|
||||
use LayLink\Protocol\FrameParser;
|
||||
use LayLink\Protocol\FrameType;
|
||||
use LayLink\Route\RouteResolver;
|
||||
use LayLink\Session\SessionManager;
|
||||
use LayLink\Session\TunnelSession;
|
||||
use Workerman\Connection\AsyncTcpConnection;
|
||||
use Workerman\Connection\AsyncUdpConnection;
|
||||
use Workerman\Connection\TcpConnection;
|
||||
use Workerman\Timer;
|
||||
use Workerman\Worker;
|
||||
|
||||
final class AgentListener
|
||||
{
|
||||
/** @var array<int, FrameParser> */
|
||||
private array $parsers = [];
|
||||
/** @var array<int, string> */
|
||||
private array $connectionNodeIds = [];
|
||||
|
||||
public function __construct(
|
||||
Worker $worker,
|
||||
private readonly NodeAuthenticator $authenticator,
|
||||
private readonly ClientAuthenticator $clientAuthenticator,
|
||||
private readonly RouteResolver $routes,
|
||||
private readonly NodeRegistry $nodes,
|
||||
private readonly SessionManager $sessions,
|
||||
private readonly AuditLogger $audit,
|
||||
) {
|
||||
$worker->onConnect = fn (TcpConnection $connection) => $this->onConnect($connection);
|
||||
$worker->onMessage = fn (TcpConnection $connection, string $data) => $this->onMessage($connection, $data);
|
||||
$worker->onClose = fn (TcpConnection $connection) => $this->onClose($connection);
|
||||
$worker->onWorkerStart = fn () => Timer::add(10, fn () => $this->sweepHeartbeats());
|
||||
}
|
||||
|
||||
private function onConnect(TcpConnection $connection): void
|
||||
{
|
||||
$connection->maxSendBufferSize = 8 * 1024 * 1024;
|
||||
$this->parsers[$connection->id] = new FrameParser();
|
||||
}
|
||||
|
||||
private function onMessage(TcpConnection $connection, string $data): void
|
||||
{
|
||||
try {
|
||||
foreach ($this->parsers[$connection->id]->push($data) as $frame) {
|
||||
$this->handleFrame($connection, $frame);
|
||||
}
|
||||
} catch (\Throwable $e) {
|
||||
$this->send($connection, new Frame(FrameType::ERROR, null, ['reason' => 'invalid_frame']));
|
||||
$connection->close();
|
||||
}
|
||||
}
|
||||
|
||||
private function handleFrame(TcpConnection $connection, Frame $frame): void
|
||||
{
|
||||
if ($frame->type === FrameType::AUTH) {
|
||||
$result = $this->authenticator->authenticate($frame->payload);
|
||||
if (!$result['ok']) {
|
||||
$this->send($connection, new Frame(FrameType::AUTH_FAIL, null, ['reason' => $result['reason']]));
|
||||
$connection->close();
|
||||
return;
|
||||
}
|
||||
|
||||
$nodeId = (string)$frame->payload['node_id'];
|
||||
$node = new NodeConnection(
|
||||
$nodeId,
|
||||
(string)$frame->payload['node_type'],
|
||||
(string)($frame->payload['node_zone'] ?? 'default'),
|
||||
$connection,
|
||||
);
|
||||
$this->nodes->register($node);
|
||||
$this->connectionNodeIds[$connection->id] = $nodeId;
|
||||
$this->send($connection, new Frame(FrameType::AUTH_OK, null, [
|
||||
'node_id' => $nodeId,
|
||||
'heartbeat_interval' => 10,
|
||||
]));
|
||||
echo "Agent online: {$nodeId}\n";
|
||||
return;
|
||||
}
|
||||
|
||||
$nodeId = $this->connectionNodeIds[$connection->id] ?? null;
|
||||
if (!is_string($nodeId) || $this->nodes->get($nodeId) === null) {
|
||||
$this->send($connection, new Frame(FrameType::ERROR, $frame->sessionId, ['reason' => 'invalid_auth']));
|
||||
$connection->close();
|
||||
return;
|
||||
}
|
||||
|
||||
if ($frame->type === FrameType::PING) {
|
||||
$node = $this->nodes->get($nodeId);
|
||||
if ($node !== null) {
|
||||
$node->lastHeartbeat = time();
|
||||
$node->activeSessions = (int)($frame->payload['active_sessions'] ?? $node->activeSessions);
|
||||
}
|
||||
$this->send($connection, new Frame(FrameType::PONG, null, ['timestamp' => time()]));
|
||||
return;
|
||||
}
|
||||
|
||||
if ($frame->type === FrameType::OPEN) {
|
||||
$this->openTargetForAgent($connection, $nodeId, $frame);
|
||||
return;
|
||||
}
|
||||
|
||||
if ($frame->type === FrameType::UDP_DATA) {
|
||||
$this->forwardUdpDatagram($connection, $nodeId, $frame);
|
||||
return;
|
||||
}
|
||||
|
||||
$session = $frame->sessionId === null ? null : $this->sessions->get($frame->sessionId);
|
||||
if ($session === null) {
|
||||
$this->send($connection, new Frame(FrameType::ERROR, $frame->sessionId, ['reason' => 'session_not_found']));
|
||||
return;
|
||||
}
|
||||
|
||||
match ($frame->type) {
|
||||
FrameType::DATA => $this->forwardDataToTarget($session, (string)($frame->payload['data'] ?? '')),
|
||||
FrameType::CLOSE => $this->closeSession($session, 'closed', null),
|
||||
default => null,
|
||||
};
|
||||
}
|
||||
|
||||
private function forwardUdpDatagram(TcpConnection $agentConnection, string $nodeId, Frame $frame): void
|
||||
{
|
||||
if ($frame->sessionId === null) {
|
||||
$this->send($agentConnection, new Frame(FrameType::ERROR, null, ['reason' => 'invalid_frame']));
|
||||
return;
|
||||
}
|
||||
|
||||
$auth = $this->clientAuthenticator->authenticate($frame->payload);
|
||||
if (!$auth['ok']) {
|
||||
$this->send($agentConnection, new Frame(FrameType::ERROR, $frame->sessionId, ['reason' => 'invalid_auth']));
|
||||
return;
|
||||
}
|
||||
|
||||
$host = (string)($frame->payload['target_host'] ?? '');
|
||||
$port = (int)($frame->payload['target_port'] ?? 0);
|
||||
$data = base64_decode((string)($frame->payload['data'] ?? ''), true);
|
||||
if ($host === '' || $port < 1 || $port > 65535 || $data === false) {
|
||||
$this->send($agentConnection, new Frame(FrameType::ERROR, $frame->sessionId, ['reason' => 'invalid_frame']));
|
||||
return;
|
||||
}
|
||||
|
||||
$decision = $this->routes->resolve((string)$auth['user_id'], $host, $port, 'udp', $frame->payload['route_hint'] ?? null);
|
||||
if (!$decision->allowed || $decision->routeType !== 'direct') {
|
||||
$this->send($agentConnection, new Frame(FrameType::ERROR, $frame->sessionId, [
|
||||
'reason' => $decision->reason ?? 'policy_denied',
|
||||
]));
|
||||
return;
|
||||
}
|
||||
|
||||
$target = new AsyncUdpConnection("udp://{$host}:{$port}");
|
||||
$target->onConnect = fn (AsyncUdpConnection $target) => $target->send($data);
|
||||
$target->onMessage = function (AsyncUdpConnection $target, string $response) use ($agentConnection, $frame, $host, $port): void {
|
||||
$this->send($agentConnection, new Frame(FrameType::UDP_DATA, $frame->sessionId, [
|
||||
'target_host' => $host,
|
||||
'target_port' => $port,
|
||||
'protocol' => 'udp',
|
||||
'data' => base64_encode($response),
|
||||
]));
|
||||
$target->close();
|
||||
};
|
||||
Timer::add(5, fn () => $target->close(), [], false);
|
||||
$target->connect();
|
||||
}
|
||||
|
||||
private function openTargetForAgent(TcpConnection $agentConnection, string $nodeId, Frame $frame): void
|
||||
{
|
||||
if ($frame->sessionId === null || $this->sessions->get($frame->sessionId) !== null) {
|
||||
$this->send($agentConnection, new Frame(FrameType::OPEN_FAIL, $frame->sessionId, ['reason' => 'invalid_frame']));
|
||||
return;
|
||||
}
|
||||
|
||||
$auth = $this->clientAuthenticator->authenticate($frame->payload);
|
||||
if (!$auth['ok']) {
|
||||
$this->rejectOpen($agentConnection, $frame, 'invalid_auth', 'anonymous', $nodeId);
|
||||
return;
|
||||
}
|
||||
|
||||
$host = (string)($frame->payload['target_host'] ?? '');
|
||||
$port = (int)($frame->payload['target_port'] ?? 0);
|
||||
$protocol = (string)($frame->payload['protocol'] ?? 'tcp');
|
||||
if ($host === '' || $port < 1 || $port > 65535 || $protocol !== 'tcp') {
|
||||
$this->rejectOpen($agentConnection, $frame, 'protocol_not_supported', (string)$auth['user_id'], $nodeId);
|
||||
return;
|
||||
}
|
||||
|
||||
$decision = $this->routes->resolve((string)$auth['user_id'], $host, $port, $protocol, $frame->payload['route_hint'] ?? null);
|
||||
if (!$decision->allowed) {
|
||||
$this->rejectOpen($agentConnection, $frame, $decision->reason ?? 'policy_denied', (string)$auth['user_id'], $nodeId, $decision->policyId);
|
||||
return;
|
||||
}
|
||||
if ($decision->routeType !== 'direct') {
|
||||
$this->rejectOpen($agentConnection, $frame, 'route_not_supported', (string)$auth['user_id'], $nodeId, $decision->policyId);
|
||||
return;
|
||||
}
|
||||
|
||||
$session = new TunnelSession(
|
||||
$frame->sessionId,
|
||||
(string)$auth['user_id'],
|
||||
(string)($frame->payload['source_ip'] ?? $agentConnection->getRemoteIp() ?? ''),
|
||||
$host,
|
||||
$port,
|
||||
$protocol,
|
||||
'direct',
|
||||
$decision->policyId,
|
||||
);
|
||||
$session->nodeId = $nodeId;
|
||||
$session->agent = $agentConnection;
|
||||
$session->state = TunnelSession::OPENING;
|
||||
$this->sessions->add($session);
|
||||
|
||||
$target = new AsyncTcpConnection("tcp://{$host}:{$port}");
|
||||
$target->maxSendBufferSize = 8 * 1024 * 1024;
|
||||
$session->target = $target;
|
||||
|
||||
$target->onConnect = function () use ($session, $agentConnection): void {
|
||||
$session->state = TunnelSession::OPEN;
|
||||
$this->send($agentConnection, new Frame(FrameType::OPEN_OK, $session->sessionId, [
|
||||
'target_host' => $session->targetHost,
|
||||
'target_port' => $session->targetPort,
|
||||
]));
|
||||
};
|
||||
$target->onMessage = function (AsyncTcpConnection $target, string $data) use ($session, $agentConnection): void {
|
||||
$session->bytesTargetToClient += strlen($data);
|
||||
$this->send($agentConnection, new Frame(FrameType::DATA, $session->sessionId, [
|
||||
'data' => base64_encode($data),
|
||||
]));
|
||||
};
|
||||
$target->onClose = fn () => $this->closeSession($session, 'closed', null);
|
||||
$target->onError = fn () => $this->failOpenSession($agentConnection, $session, 'target_connection_refused');
|
||||
$target->connect();
|
||||
}
|
||||
|
||||
private function forwardDataToTarget(TunnelSession $session, string $encoded): void
|
||||
{
|
||||
$data = base64_decode($encoded, true);
|
||||
if ($data === false) {
|
||||
$this->closeSession($session, 'failed', 'invalid_frame');
|
||||
return;
|
||||
}
|
||||
|
||||
$session->bytesClientToTarget += strlen($data);
|
||||
$session->target?->send($data);
|
||||
}
|
||||
|
||||
private function failOpenSession(TcpConnection $agentConnection, TunnelSession $session, string $reason): void
|
||||
{
|
||||
$this->send($agentConnection, new Frame(FrameType::OPEN_FAIL, $session->sessionId, ['reason' => $reason]));
|
||||
$this->closeSession($session, 'failed', $reason);
|
||||
}
|
||||
|
||||
private function rejectOpen(TcpConnection $agentConnection, Frame $frame, string $reason, string $userId, string $nodeId, ?string $policyId = null): void
|
||||
{
|
||||
$this->send($agentConnection, new Frame(FrameType::OPEN_FAIL, $frame->sessionId, ['reason' => $reason]));
|
||||
$this->audit->write([
|
||||
'session_id' => $frame->sessionId,
|
||||
'user_id' => $userId,
|
||||
'source_ip' => (string)($frame->payload['source_ip'] ?? $agentConnection->getRemoteIp() ?? ''),
|
||||
'target_host' => (string)($frame->payload['target_host'] ?? ''),
|
||||
'target_port' => (int)($frame->payload['target_port'] ?? 0),
|
||||
'protocol' => (string)($frame->payload['protocol'] ?? 'tcp'),
|
||||
'route_type' => 'reject',
|
||||
'node_id' => $nodeId,
|
||||
'policy_id' => $policyId,
|
||||
'start_time' => date(DATE_ATOM),
|
||||
'duration_ms' => 0,
|
||||
'bytes_client_to_target' => 0,
|
||||
'bytes_target_to_client' => 0,
|
||||
'result' => 'failed',
|
||||
'failure_reason' => $reason,
|
||||
]);
|
||||
}
|
||||
|
||||
public function closeSession(TunnelSession $session, string $result, ?string $reason): void
|
||||
{
|
||||
if ($session->state === TunnelSession::CLOSED) {
|
||||
return;
|
||||
}
|
||||
|
||||
$session->state = TunnelSession::CLOSED;
|
||||
$this->sessions->remove($session->sessionId);
|
||||
|
||||
if ($session->agent !== null) {
|
||||
$this->send($session->agent, new Frame(FrameType::CLOSE, $session->sessionId, ['reason' => $reason ?? $result]));
|
||||
}
|
||||
$session->client?->close();
|
||||
$session->target?->close();
|
||||
|
||||
$endMs = (int)floor(microtime(true) * 1000);
|
||||
$this->audit->write([
|
||||
'session_id' => $session->sessionId,
|
||||
'user_id' => $session->userId,
|
||||
'source_ip' => $session->sourceIp,
|
||||
'target_host' => $session->targetHost,
|
||||
'target_port' => $session->targetPort,
|
||||
'protocol' => $session->protocol,
|
||||
'route_type' => $session->routeType,
|
||||
'node_id' => $session->nodeId,
|
||||
'policy_id' => $session->policyId,
|
||||
'start_time' => $session->startTime,
|
||||
'duration_ms' => $endMs - $session->startedAtMs,
|
||||
'bytes_client_to_target' => $session->bytesClientToTarget,
|
||||
'bytes_target_to_client' => $session->bytesTargetToClient,
|
||||
'result' => $result,
|
||||
'failure_reason' => $reason,
|
||||
]);
|
||||
}
|
||||
|
||||
private function onClose(TcpConnection $connection): void
|
||||
{
|
||||
unset($this->parsers[$connection->id]);
|
||||
unset($this->connectionNodeIds[$connection->id]);
|
||||
$node = $this->nodes->unregisterByConnection($connection);
|
||||
foreach ($this->sessions->all() as $session) {
|
||||
if ($session->agent === $connection) {
|
||||
$this->closeSession($session, 'failed', 'node_offline');
|
||||
}
|
||||
}
|
||||
if ($node === null) {
|
||||
return;
|
||||
}
|
||||
|
||||
echo "Agent offline: {$node->nodeId}\n";
|
||||
}
|
||||
|
||||
private function sweepHeartbeats(): void
|
||||
{
|
||||
foreach ($this->nodes->all() as $node) {
|
||||
if (time() - $node->lastHeartbeat > 30) {
|
||||
$node->connection->close();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private function send(TcpConnection $connection, Frame $frame): void
|
||||
{
|
||||
$connection->send(FrameCodec::encode($frame));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
<?php
|
||||
|
||||
declare(strict_types=1);
|
||||
|
||||
namespace LayLink\Server;
|
||||
|
||||
use LayLink\Audit\AuditLogger;
|
||||
use LayLink\Auth\ClientAuthenticator;
|
||||
use LayLink\Auth\NodeAuthenticator;
|
||||
use LayLink\Auth\PolicyChecker;
|
||||
use LayLink\Node\NodeRegistry;
|
||||
use LayLink\Route\RouteResolver;
|
||||
use LayLink\Session\SessionManager;
|
||||
use Workerman\Worker;
|
||||
|
||||
final class PopServer
|
||||
{
|
||||
public readonly NodeRegistry $nodes;
|
||||
public readonly SessionManager $sessions;
|
||||
public readonly AuditLogger $audit;
|
||||
|
||||
public function __construct(
|
||||
private readonly string $agentListen,
|
||||
private readonly array $nodeConfig,
|
||||
private readonly array $policies,
|
||||
private readonly array $allowedAgentTransports,
|
||||
string $auditLog,
|
||||
) {
|
||||
$this->nodes = new NodeRegistry();
|
||||
$this->sessions = new SessionManager();
|
||||
$this->audit = new AuditLogger($auditLog);
|
||||
}
|
||||
|
||||
public function boot(): void
|
||||
{
|
||||
$agentWorker = new Worker('tcp://' . $this->agentListen);
|
||||
$agentWorker->name = 'laylink-pop-agent-listener';
|
||||
$agentWorker->count = 1;
|
||||
new AgentListener(
|
||||
$agentWorker,
|
||||
new NodeAuthenticator($this->nodeConfig, $this->allowedAgentTransports),
|
||||
new ClientAuthenticator(),
|
||||
new RouteResolver(new PolicyChecker($this->policies), $this->nodes),
|
||||
$this->nodes,
|
||||
$this->sessions,
|
||||
$this->audit,
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user