Lines 91.35% 74 / 81
Functions and Methods 62.50% 5 / 8
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
ControlServer 91.35% 74 / 81 62.50% 5 / 8 27.47 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 start 76.92% 10 / 13 0.00% 0 / 1 3.11
 stop 100.00% 9 / 9 100.00% 1 / 1 3
 acceptClients 78.57% 11 / 14 0.00% 0 / 1 6.35
 serveClient 100.00% 13 / 13 100.00% 1 / 1 4
 awaitHello 100.00% 17 / 17 100.00% 1 / 1 5
 answerRequest 100.00% 9 / 9 100.00% 1 / 1 3
 prepareBindAddress 80.00% 4 / 5 0.00% 0 / 1 2.03
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Runtime\Control\Server;
6
7use Amp\Socket;
8use Amp\Socket\ResourceServerSocket;
9use Amp\Socket\SocketAddress;
10use Amp\Socket\UnixAddress;
11use Psr\Log\LoggerInterface;
12use SensitiveParameter;
13use Stewart\Contracts\Exception\StewartException;
14use Stewart\Contracts\Time\Duration;
15use Stewart\Runtime\Broker\ControlPlane;
16use Stewart\Runtime\Config\ControlAddress;
17use Stewart\Runtime\Config\ProjectRoot;
18use Stewart\Runtime\Config\UnixControlAddress;
19use Stewart\Runtime\Control\Protocol\Codec\FrameCodec;
20use Stewart\Runtime\Control\Protocol\ControlProtocol;
21use Stewart\Runtime\Control\Protocol\Frame\Bye;
22use Stewart\Runtime\Control\Protocol\Frame\Hello;
23use Stewart\Runtime\Control\Protocol\Frame\Rejected;
24use Stewart\Runtime\Control\Protocol\Frame\RequestFailed;
25use Stewart\Runtime\Control\Protocol\Frame\ServerFrame;
26use Stewart\Runtime\Control\Protocol\Frame\Welcome;
27use Stewart\Runtime\Control\Request\ControlRequestDispatcher;
28use Stewart\Runtime\Exception\ConfigurationException;
29use Stewart\Runtime\Exception\ControlError;
30use Stewart\Runtime\Exception\ControlException;
31use Stewart\Support\Time\Deadlines;
32use Throwable;
33
34use function Amp\async;
35
36final class ControlServer implements ControlPlane
37{
38    private const int LINE_LIMIT = 65536;
39
40    private const int MAX_CLIENTS = 8;
41
42    private ?ResourceServerSocket $server = null;
43
44    /** @var array<int, ClientSession> */
45    private array $sessions = [];
46
47    private int $nextClientId = 0;
48
49    private ?BoundSocketFile $boundSocketFile = null;
50
51    private readonly Duration $sessionTimeout;
52
53    public function __construct(
54        private readonly ControlAddress $address,
55        #[SensitiveParameter]
56        private readonly string $token,
57        private readonly ControlRequestDispatcher $requests,
58        private readonly Deadlines $deadlines,
59        private readonly FrameCodec $codec,
60        private readonly LoggerInterface $logger,
61        private readonly UnixSocketFile $socketFile,
62        private readonly ProjectRoot $projectRoot,
63        ?Duration $sessionTimeout = null,
64    ) {
65        $this->sessionTimeout = $sessionTimeout ?? Duration::seconds(5);
66    }
67
68    /** @throws ControlException|ConfigurationException */
69    public function start(): void
70    {
71        $bindAddress = $this->prepareBindAddress();
72        // umask instead of chmod: the socket is never readable by other users, not even briefly.
73        $mask = umask(0o177);
74
75        try {
76            $this->server = Socket\listen($bindAddress);
77        } finally {
78            umask($mask);
79        }
80
81        if ($bindAddress instanceof UnixAddress) {
82            $this->boundSocketFile = $this->socketFile->identifyBoundSocket($bindAddress->toString());
83        }
84
85        $this->server->unreference();
86
87        async($this->acceptClients(...))->ignore();
88
89        $this->logger->info('Control socket listening', ['address' => (string) $this->address]);
90
91        if (!$this->address->isLoopback()) {
92            $this->logger->warning('The control socket accepts connections from other hosts; anyone with the token can read the runtime snapshot and pause apps', [
93                'address' => (string) $this->address,
94            ]);
95        }
96    }
97
98    public function stop(): void
99    {
100        $server = $this->server;
101        $this->server = null;
102        $server?->close();
103
104        foreach ($this->sessions as $session) {
105            $session->close();
106        }
107
108        $this->sessions = [];
109
110        if ($this->boundSocketFile !== null) {
111            $this->socketFile->removeIfStillBound($this->boundSocketFile);
112            $this->boundSocketFile = null;
113        }
114    }
115
116    private function acceptClients(): void
117    {
118        $server = $this->server;
119
120        if ($server === null) {
121            return;
122        }
123
124        try {
125            while (($socket = $server->accept()) !== null) {
126                if (\count($this->sessions) >= self::MAX_CLIENTS) {
127                    $this->logger->warning('Control client refused: too many clients', ['limit' => self::MAX_CLIENTS]);
128                    $socket->close();
129
130                    continue;
131                }
132
133                $session = new ClientSession(++$this->nextClientId, $socket, $this->codec, $this->deadlines->timeout($this->sessionTimeout));
134                $this->sessions[$session->id] = $session;
135                async($this->serveClient(...), $session)->ignore();
136            }
137        } catch (Throwable $e) {
138            if ($this->server !== null) {
139                $this->logger->error('Control socket stopped accepting connections', ['exception' => $e]);
140            }
141        }
142    }
143
144    private function serveClient(ClientSession $session): void
145    {
146        try {
147            $hello = $this->awaitHello($session);
148
149            if ($hello === null) {
150                return;
151            }
152
153            $session->send(new Welcome(ControlProtocol::VERSION));
154            $session->send($this->answerRequest($session));
155            $session->send(new Bye('request answered'));
156            $this->logger->debug('Answered a control client', ['client' => $hello->client, 'peer' => $session->describePeer()]);
157        } catch (Throwable $e) {
158            if ($session->hasExpired()) {
159                $this->logger->warning('Dropped a control client that did not finish in time', ['peer' => $session->describePeer(), 'timeout' => (string) $this->sessionTimeout]);
160            } else {
161                $this->logger->warning('Could not answer a control client', ['peer' => $session->describePeer(), 'exception' => $e]);
162            }
163        } finally {
164            unset($this->sessions[$session->id]);
165            $session->close();
166        }
167    }
168
169    private function awaitHello(ClientSession $session): ?Hello
170    {
171        try {
172            $frame = $this->codec->decodeClientFrame($session->readLine(self::LINE_LIMIT));
173        } catch (Throwable $e) {
174            $this->logger->warning('Refused a control connection that did not say hello', ['peer' => $session->describePeer(), 'exception' => $e]);
175
176            return null;
177        }
178
179        if (!$frame instanceof Hello || !hash_equals($this->token, $frame->token)) {
180            $this->logger->warning('Refused a control connection with the wrong token', ['peer' => $session->describePeer()]);
181            $session->send(new Rejected('the token does not match control.token'));
182
183            return null;
184        }
185
186        if ($frame->protocol !== ControlProtocol::VERSION) {
187            $this->logger->warning('Refused a control client that speaks another protocol', ['peer' => $session->describePeer(), 'protocol' => $frame->protocol]);
188            $session->send(new Rejected(\sprintf(
189                'this daemon speaks control protocol %d and the client speaks %d; use the same Stewart version on both sides',
190                ControlProtocol::VERSION,
191                $frame->protocol,
192            )));
193
194            return null;
195        }
196
197        return $frame;
198    }
199
200    private function answerRequest(ClientSession $session): ServerFrame
201    {
202        $line = $session->readLine(self::LINE_LIMIT);
203
204        try {
205            return $this->requests->answerRequest($this->codec->decodeClientFrame($line));
206        } catch (StewartException $e) {
207            $this->logger->info('Refused a control request', ['peer' => $session->describePeer(), 'reason' => $e->reason->value]);
208
209            return new RequestFailed((string) $e->reason->value, $e->getMessage());
210        } catch (Throwable $e) {
211            $this->logger->warning('Could not answer a control request', ['peer' => $session->describePeer(), 'exception' => $e]);
212            $unanswerable = ControlException::createForReason(ControlError::RequestUnanswerable, $e);
213
214            return new RequestFailed($unanswerable->reason->value, $unanswerable->getMessage());
215        }
216    }
217
218    /** @throws ControlException|ConfigurationException */
219    private function prepareBindAddress(): SocketAddress|string
220    {
221        if (!$this->address instanceof UnixControlAddress) {
222            return $this->address->resolveSocketAddress($this->projectRoot);
223        }
224
225        $path = $this->address->resolveSocketPath($this->projectRoot);
226        $this->socketFile->prepareForListening($path);
227
228        return new UnixAddress($path);
229    }
230}