Skip to content
rachel/workerd
File

Blob: src/workerd/api/sockets.h

cpp314 lines
1// Copyright (c) 2017-2022 Cloudflare, Inc.
2// Licensed under the Apache 2.0 license found in the LICENSE file or at:
3//     https://opensource.org/licenses/Apache-2.0
4 
5#pragma once
6 
7#include <workerd/api/streams/readable.h>
8#include <workerd/api/streams/writable.h>
9#include <workerd/jsg/jsg.h>
10#include <workerd/jsg/modules-new.h>
11#include <workerd/jsg/url.h>
12 
13namespace workerd::api {
14 
15class Fetcher;
16 
17enum class SecureTransportKind {
18  // plain-text
19  OFF,
20  // plain-text at first, with `startTls` available to upgrade at a later time
21  STARTTLS,
22  // TLS enabled immediately
23  ON,
24};
25 
26struct SocketAddress {
27  kj::String hostname;
28  uint16_t port;
29  JSG_STRUCT(hostname, port);
30};
31 
32struct SocketInfo {
33  jsg::Optional<kj::String> remoteAddress;
34 
35  // The local address — i.e. the address on this side of the socket. For outbound sockets created
36  // via `connect()`, we don't have a useful value to provide and leave it empty. For inbound
37  // sockets delivered to a worker's `connect(socket)` handler, this is populated with the CONNECT
38  // authority (the "host:port" string the caller passed to `fetcher.connect(...)`), since from the
39  // handler's perspective that is the address the peer asked to connect to on this end.
40  jsg::Optional<kj::String> localAddress;
41  JSG_STRUCT(remoteAddress, localAddress);
42};
43 
44using AnySocketAddress = kj::OneOf<SocketAddress, kj::String>;
45 
46struct SocketOptions {
47  jsg::Optional<kj::String> secureTransport;
48  bool allowHalfOpen = false;
49  jsg::Optional<uint64_t> highWaterMark;
50  JSG_STRUCT(secureTransport, allowHalfOpen, highWaterMark);
51  JSG_MEMORY_INFO(SocketOptions) {
52    tracker.trackField("secureTransport", secureTransport);
53  }
54};
55 
56struct TlsOptions {
57  jsg::Optional<kj::String> expectedServerHostname;
58  JSG_STRUCT(expectedServerHostname);
59};
60 
61class Socket: public jsg::Object {
62 public:
63  Socket(jsg::Lock& js,
64      IoContext& context,
65      kj::Own<kj::RefcountedWrapper<kj::Own<kj::AsyncIoStream>>> connectionStream,
66      kj::Maybe<kj::String> remoteAddress,
67      kj::Maybe<kj::String> localAddress,
68      jsg::Ref<ReadableStream> readableParam,
69      jsg::Ref<WritableStream> writable,
70      jsg::PromiseResolverPair<void> closedPrPair,
71      kj::Promise<void> watchForDisconnectTask,
72      jsg::Optional<SocketOptions> options,
73      kj::Own<kj::TlsStarterCallback> tlsStarter,
74      SecureTransportKind secureTransport,
75      kj::Maybe<kj::String> domain,
76      bool isDefaultFetchPort,
77      jsg::PromiseResolverPair<SocketInfo> openedPrPair)
78      : connectionData(context.addObject(kj::heap<ConnectionData>(
79            kj::mv(tlsStarter), kj::mv(connectionStream), kj::mv(watchForDisconnectTask)))),
80        readable(kj::mv(readableParam)),
81        writable(kj::mv(writable)),
82        closedResolver(kj::mv(closedPrPair.resolver)),
83        closedPromiseCopy(closedPrPair.promise.whenResolved(js)),
84        closedPromise(kj::mv(closedPrPair.promise)),
85        options(kj::mv(options)),
86        remoteAddress(kj::mv(remoteAddress)),
87        localAddress(kj::mv(localAddress)),
88        secureTransport(secureTransport),
89        domain(kj::mv(domain)),
90        isDefaultFetchPort(isDefaultFetchPort),
91        openedResolver(kj::mv(openedPrPair.resolver)),
92        openedPromiseCopy(openedPrPair.promise.whenResolved(js)),
93        openedPromise(kj::mv(openedPrPair.promise)) {};
94 
95  jsg::Ref<ReadableStream> getReadable() {
96    return readable.addRef();
97  }
98  jsg::Ref<WritableStream> getWritable() {
99    return writable.addRef();
100  }
101  jsg::MemoizedIdentity<jsg::Promise<void>>& getClosed() {
102    return closedPromise;
103  }
104  jsg::MemoizedIdentity<jsg::Promise<SocketInfo>>& getOpened() {
105    return openedPromise;
106  }
107 
108  bool getUpgraded() const {
109    return upgraded;
110  }
111 
112  kj::StringPtr getSecureTransport() const {
113    switch (secureTransport) {
114      case SecureTransportKind::OFF:
115        return "off"_kj;
116      case SecureTransportKind::STARTTLS:
117        return "starttls"_kj;
118      case SecureTransportKind::ON:
119        return "on"_kj;
120    }
121  }
122 
123  // Takes ownership of the underlying connection stream, detaching the readable and writable streams.
124  // This is a destructive operation that renders the Socket unusable for further I/O operations.
125  kj::Own<kj::AsyncIoStream> takeConnectionStream(jsg::Lock& js);
126 
127  // Closes the socket connection.
128  //
129  // The closure is only performed after the socket connection is properly
130  // established through any configured proxy. This method also flushes the writable stream prior to
131  // closing.
132  jsg::Promise<void> close(jsg::Lock& js);
133 
134  // Flushes write buffers then performs a TLS handshake on the current Socket connection.
135  // The current `Socket` instance is closed and its readable/writable instances are also closed.
136  // All new operations should be performed on the new `Socket` instance.
137  jsg::Ref<Socket> startTls(jsg::Lock& js, jsg::Optional<TlsOptions> options);
138 
139  // Sets up relevant callbacks to handle the case when the proxy rejects our connection.
140  // The first variant is useful for connections established using HTTP connect. The latter is for
141  // connections established any other way, where the lack of an exception indicates we connected
142  // successfully.
143  void handleProxyStatus(jsg::Lock& js, kj::Promise<kj::HttpClient::ConnectRequest::Status> status);
144 
145  // Sets up relevant callbacks to handle the case when the proxy rejects our connection.
146  // The first variant is useful for connections established using HTTP connect. The latter is for
147  // connections established any other way, where the lack of an exception indicates we connected
148  // successfully.
149  void handleProxyStatus(jsg::Lock& js, kj::Promise<kj::Maybe<kj::Exception>> status);
150 
151  void handleReadableEof(jsg::Lock& js, jsg::Promise<void> onEof);
152  // Sets up relevant callbacks to handle the case when the readable stream reaches EOF.
153 
154  JSG_RESOURCE_TYPE(Socket) {
155    JSG_READONLY_PROTOTYPE_PROPERTY(readable, getReadable);
156    JSG_READONLY_PROTOTYPE_PROPERTY(writable, getWritable);
157    JSG_READONLY_PROTOTYPE_PROPERTY(closed, getClosed);
158    JSG_READONLY_PROTOTYPE_PROPERTY(opened, getOpened);
159    JSG_READONLY_PROTOTYPE_PROPERTY(upgraded, getUpgraded);
160    JSG_READONLY_PROTOTYPE_PROPERTY(secureTransport, getSecureTransport);
161    JSG_METHOD(close);
162    JSG_METHOD(startTls);
163 
164    JSG_TS_OVERRIDE({
165      get secureTransport(): 'on' | 'off' | 'starttls';
166    });
167  }
168 
169  void visitForMemoryInfo(jsg::MemoryTracker& tracker) const {
170    tracker.trackFieldWithSize("connectionData", sizeof(IoOwn<ConnectionData>));
171    tracker.trackField("readable", readable);
172    tracker.trackField("writable", writable);
173    tracker.trackField("closedResolver", closedResolver);
174    tracker.trackField("closedPromiseCopy", closedPromiseCopy);
175    tracker.trackField("closedPromise", closedPromise);
176    tracker.trackField("options", options);
177    tracker.trackField("domain", domain);
178    tracker.trackField("openedResolver", openedResolver);
179    tracker.trackField("openedPromiseCopy", openedPromiseCopy);
180    tracker.trackField("openedPromise", openedPromise);
181  }
182 
183 private:
184  struct ConnectionData {
185    kj::Own<kj::RefcountedWrapper<kj::Own<kj::AsyncIoStream>>> connectionStream;
186    kj::Maybe<kj::Promise<void>> watchForDisconnectTask;
187    // tlsStarter must be declared after connectionStream so that it is destroyed first,
188    // since it holds a reference that keeps the connection alive.
189    kj::Own<kj::TlsStarterCallback> tlsStarter;
190    ConnectionData(kj::Own<kj::TlsStarterCallback> tlsStarter,
191        kj::Own<kj::RefcountedWrapper<kj::Own<kj::AsyncIoStream>>> connStream,
192        kj::Promise<void> disconnectTask)
193        : connectionStream(kj::mv(connStream)),
194          watchForDisconnectTask(kj::mv(disconnectTask)),
195          tlsStarter(kj::mv(tlsStarter)) {}
196  };
197  kj::Maybe<IoOwn<ConnectionData>> connectionData;
198 
199  jsg::Ref<ReadableStream> readable;
200  jsg::Ref<WritableStream> writable;
201  // This fulfiller is used to resolve the `closedPromise` below.
202  jsg::Promise<void>::Resolver closedResolver;
203  // Copy kept so that it can be returned from `close`.
204  jsg::Promise<void> closedPromiseCopy;
205  // Memoized copy that is returned by the `closed` attribute.
206  jsg::MemoizedIdentity<jsg::Promise<void>> closedPromise;
207  jsg::Optional<SocketOptions> options;
208  kj::Maybe<kj::String> remoteAddress;
209  kj::Maybe<kj::String> localAddress;
210  // Set to true when the socket is upgraded to a secure one.
211  bool upgraded = false;
212  SecureTransportKind secureTransport;
213  // The domain/ip this socket is connected to. Used for startTls.
214  kj::Maybe<kj::String> domain;
215  // Whether the port this socket connected to is 80/443. Used for nicer errors.
216  bool isDefaultFetchPort;
217  // This fulfiller is used to resolve the `openedPromise` below.
218  jsg::Promise<SocketInfo>::Resolver openedResolver;
219  // Copy kept so that it can be used in `close`.
220  jsg::Promise<void> openedPromiseCopy;
221  jsg::MemoizedIdentity<jsg::Promise<SocketInfo>> openedPromise;
222  // Used to keep track of a pending `close` operation on the socket.
223  bool isClosing = false;
224 
225  kj::Promise<kj::Own<kj::AsyncIoStream>> processConnection();
226  jsg::Promise<void> maybeCloseWriteSide(jsg::Lock& js);
227  jsg::Promise<void> closeImplOld(jsg::Lock& js);
228  jsg::Promise<void> closeImplNew(jsg::Lock& js);
229 
230  // Helper method for handleProxyStatus implementations.
231  void handleProxyError(jsg::Lock& js, kj::Exception e);
232 
233  void resolveFulfiller(jsg::Lock& js, kj::Maybe<kj::Exception> maybeErr) {
234    KJ_IF_SOME(err, maybeErr) {
235      closedResolver.reject(js, err.clone());
236    } else {
237      closedResolver.resolve(js);
238    }
239  };
240 
241  void errorHandler(jsg::Lock& js, jsg::Value err) {
242    auto jsException = err.getHandle(js);
243    resolveFulfiller(js, jsg::createTunneledException(js.v8Isolate, jsException));
244  };
245 
246  void visitForGc(jsg::GcVisitor& visitor) {
247    visitor.visit(readable, writable, closedResolver, closedPromiseCopy, closedPromise,
248        openedResolver, openedPromiseCopy, openedPromise);
249  }
250};
251 
252jsg::Ref<Socket> setupSocket(jsg::Lock& js,
253    kj::Own<kj::AsyncIoStream> connection,
254    kj::Maybe<kj::String> remoteAddress,
255    kj::Maybe<kj::String> localAddress,
256    jsg::Optional<SocketOptions> options,
257    kj::Own<kj::TlsStarterCallback> tlsStarter,
258    SecureTransportKind secureTransport,
259    kj::Maybe<kj::String> domain,
260    bool isDefaultFetchPort,
261    kj::Maybe<jsg::PromiseResolverPair<SocketInfo>> maybeOpenedPrPair);
262 
263jsg::Ref<Socket> connectImplNoOutputLock(jsg::Lock& js,
264    kj::Maybe<jsg::Ref<Fetcher>> fetcher,
265    AnySocketAddress address,
266    jsg::Optional<SocketOptions> options);
267 
268jsg::Ref<Socket> connectImpl(jsg::Lock& js,
269    kj::Maybe<jsg::Ref<Fetcher>> fetcher,
270    AnySocketAddress address,
271    jsg::Optional<SocketOptions> options);
272 
273class SocketsModule final: public jsg::Object {
274 public:
275  SocketsModule() = default;
276  SocketsModule(jsg::Lock&, const jsg::Url&) {}
277 
278  jsg::Ref<Socket> connect(
279      jsg::Lock& js, AnySocketAddress address, jsg::Optional<SocketOptions> options);
280 
281  // Creates a Fetcher from a Socket that can perform HTTP requests over the socket connection
282  jsg::Promise<jsg::Ref<Fetcher>> internalNewHttpClient(jsg::Lock& js, jsg::Ref<Socket> socket);
283 
284  JSG_RESOURCE_TYPE(SocketsModule, CompatibilityFlags::Reader flags) {
285    JSG_METHOD(connect);
286 
287    if (flags.getWorkerdExperimental()) {
288      JSG_METHOD(internalNewHttpClient);
289    }
290  }
291};
292 
293template <class Registry>
294void registerSocketsModule(Registry& registry, auto featureFlags) {
295  registry.template addBuiltinModule<SocketsModule>(
296      "cloudflare-internal:sockets", workerd::jsg::ModuleRegistry::Type::INTERNAL);
297}
298 
299template <typename TypeWrapper>
300kj::Own<jsg::modules::ModuleBundle> getInternalSocketModuleBundle(auto featureFlags) {
301  jsg::modules::ModuleBundle::BuiltinBuilder builder(
302      jsg::modules::ModuleBundle::BuiltinBuilder::Type::BUILTIN_ONLY);
303  static const auto kSpecifier = "cloudflare-internal:sockets"_url;
304  builder.addObject<SocketsModule, TypeWrapper>(kSpecifier);
305  return builder.finish();
306}
307 
308#define EW_SOCKETS_ISOLATE_TYPES                                                                   \
309  api::Socket, api::SocketOptions, api::SocketAddress, api::TlsOptions, api::SocketsModule,        \
310      api::SocketInfo
311 
312// The list of sockets.h types that are added to worker.c++'s JSG_DECLARE_ISOLATE_TYPE
313}  // namespace workerd::api