meekConn.go 48 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455
  1. /*
  2. * Copyright (c) 2015, Psiphon Inc.
  3. * All rights reserved.
  4. *
  5. * This program is free software: you can redistribute it and/or modify
  6. * it under the terms of the GNU General Public License as published by
  7. * the Free Software Foundation, either version 3 of the License, or
  8. * (at your option) any later version.
  9. *
  10. * This program is distributed in the hope that it will be useful,
  11. * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13. * GNU General Public License for more details.
  14. *
  15. * You should have received a copy of the GNU General Public License
  16. * along with this program. If not, see <http://www.gnu.org/licenses/>.
  17. *
  18. */
  19. package psiphon
  20. import (
  21. "bytes"
  22. "context"
  23. "crypto/rand"
  24. "crypto/tls"
  25. "encoding/base64"
  26. "encoding/json"
  27. "errors"
  28. "fmt"
  29. "io"
  30. "io/ioutil"
  31. "net"
  32. "net/http"
  33. "net/url"
  34. "strings"
  35. "sync"
  36. "sync/atomic"
  37. "time"
  38. "github.com/Psiphon-Labs/goarista/monotime"
  39. "github.com/Psiphon-Labs/net/http2"
  40. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common"
  41. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/crypto/nacl/box"
  42. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/obfuscator"
  43. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/parameters"
  44. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/prng"
  45. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/protocol"
  46. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/quic"
  47. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/upstreamproxy"
  48. )
  49. // MeekConn is based on meek-client.go from Tor and Psiphon:
  50. //
  51. // https://gitweb.torproject.org/pluggable-transports/meek.git/blob/HEAD:/meek-client/meek-client.go
  52. // CC0 1.0 Universal
  53. //
  54. // https://bitbucket.org/psiphon/psiphon-circumvention-system/src/default/go/meek-client/meek-client.go
  55. const (
  56. MEEK_PROTOCOL_VERSION = 3
  57. MEEK_MAX_REQUEST_PAYLOAD_LENGTH = 65536
  58. )
  59. // MeekConfig specifies the behavior of a MeekConn
  60. type MeekConfig struct {
  61. // DiagnosticID is the server ID to record in any diagnostics notices.
  62. DiagnosticID string
  63. // ClientParameters is the active set of client parameters to use
  64. // for the meek dial.
  65. ClientParameters *parameters.ClientParameters
  66. // DialAddress is the actual network address to dial to establish a
  67. // connection to the meek server. This may be either a fronted or
  68. // direct address. The address must be in the form "host:port",
  69. // where host may be a domain name or IP address.
  70. DialAddress string
  71. // UseQUIC indicates whether to use HTTP/2 over QUIC.
  72. UseQUIC bool
  73. // QUICVersion indicates which QUIC version to use.
  74. QUICVersion string
  75. // UseHTTPS indicates whether to use HTTPS (true) or HTTP (false).
  76. // Ignored when UseQUIC is true.
  77. UseHTTPS bool
  78. // TLSProfile specifies the value for CustomTLSConfig.TLSProfile for all
  79. // underlying TLS connections created by this meek connection.
  80. TLSProfile string
  81. // NoDefaultTLSSessionID specifies the value for
  82. // CustomTLSConfig.NoDefaultTLSSessionID for all underlying TLS connections
  83. // created by this meek connection.
  84. NoDefaultTLSSessionID bool
  85. // RandomizedTLSProfileSeed specifies the value for
  86. // CustomTLSConfig.RandomizedTLSProfileSeed for all underlying TLS
  87. // connections created by this meek connection.
  88. RandomizedTLSProfileSeed *prng.Seed
  89. // UseObfuscatedSessionTickets indicates whether to use obfuscated
  90. // session tickets. Assumes UseHTTPS is true.
  91. UseObfuscatedSessionTickets bool
  92. // SNIServerName is the value to place in the TLS/QUIC SNI server_name
  93. // field when HTTPS or QUIC is used.
  94. SNIServerName string
  95. // HostHeader is the value to place in the HTTP request Host header.
  96. HostHeader string
  97. // TransformedHostName records whether a hostname transformation is
  98. // in effect. This value is used for stats reporting.
  99. TransformedHostName bool
  100. // ClientTunnelProtocol is the protocol the client is using. It's
  101. // included in the meek cookie for optional use by the server, in
  102. // cases where the server cannot unambiguously determine the
  103. // tunnel protocol.
  104. // ClientTunnelProtocol is used when selecting tactics targeted at
  105. // specific protocols.
  106. ClientTunnelProtocol string
  107. // RoundTripperOnly sets the MeekConn to operate in round tripper
  108. // mode, which is used for untunneled tactics requests. In this
  109. // mode, a connection is established to the meek server as usual,
  110. // but instead of relaying tunnel traffic, the RoundTrip function
  111. // may be used to make requests. In this mode, no relay resources
  112. // incuding buffers are allocated.
  113. RoundTripperOnly bool
  114. // NetworkLatencyMultiplier specifies a custom network latency multiplier to
  115. // apply to client parameters used by this meek connection.
  116. NetworkLatencyMultiplier float64
  117. // The following values are used to create the obfuscated meek cookie.
  118. MeekCookieEncryptionPublicKey string
  119. MeekObfuscatedKey string
  120. MeekObfuscatorPaddingSeed *prng.Seed
  121. }
  122. // MeekConn is a network connection that tunnels TCP over HTTP and supports "fronting". Meek sends
  123. // client->server flow in HTTP request bodies and receives server->client flow in HTTP response bodies.
  124. // Polling is used to achieve full duplex TCP.
  125. //
  126. // Fronting is an obfuscation technique in which the connection
  127. // to a web server, typically a CDN, is indistinguishable from any other HTTPS connection to the generic
  128. // "fronting domain" -- the HTTP Host header is used to route the requests to the actual destination.
  129. // See https://trac.torproject.org/projects/tor/wiki/doc/meek for more details.
  130. //
  131. // MeekConn also operates in unfronted mode, in which plain HTTP connections are made without routing
  132. // through a CDN.
  133. type MeekConn struct {
  134. clientParameters *parameters.ClientParameters
  135. networkLatencyMultiplier float64
  136. isQUIC bool
  137. url *url.URL
  138. additionalHeaders http.Header
  139. cookie *http.Cookie
  140. cookieSize int
  141. limitRequestPayloadLength int
  142. redialTLSProbability float64
  143. cachedTLSDialer *cachedTLSDialer
  144. transport transporter
  145. mutex sync.Mutex
  146. isClosed bool
  147. runCtx context.Context
  148. stopRunning context.CancelFunc
  149. relayWaitGroup *sync.WaitGroup
  150. // For round tripper mode
  151. roundTripperOnly bool
  152. meekCookieEncryptionPublicKey string
  153. meekObfuscatedKey string
  154. meekObfuscatorPaddingSeed *prng.Seed
  155. clientTunnelProtocol string
  156. // For relay mode
  157. fullReceiveBufferLength int
  158. readPayloadChunkLength int
  159. emptyReceiveBuffer chan *bytes.Buffer
  160. partialReceiveBuffer chan *bytes.Buffer
  161. fullReceiveBuffer chan *bytes.Buffer
  162. emptySendBuffer chan *bytes.Buffer
  163. partialSendBuffer chan *bytes.Buffer
  164. fullSendBuffer chan *bytes.Buffer
  165. }
  166. func (conn *MeekConn) getCustomClientParameters() parameters.ClientParametersAccessor {
  167. return conn.clientParameters.GetCustom(conn.networkLatencyMultiplier)
  168. }
  169. // transporter is implemented by both http.Transport and upstreamproxy.ProxyAuthTransport.
  170. type transporter interface {
  171. CloseIdleConnections()
  172. RoundTrip(req *http.Request) (resp *http.Response, err error)
  173. }
  174. // DialMeek returns an initialized meek connection. A meek connection is
  175. // an HTTP session which does not depend on an underlying socket connection (although
  176. // persistent HTTP connections are used for performance). This function does not
  177. // wait for the connection to be "established" before returning. A goroutine
  178. // is spawned which will eventually start HTTP polling.
  179. // When frontingAddress is not "", fronting is used. This option assumes caller has
  180. // already checked server entry capabilities.
  181. func DialMeek(
  182. ctx context.Context,
  183. meekConfig *MeekConfig,
  184. dialConfig *DialConfig) (meek *MeekConn, err error) {
  185. runCtx, stopRunning := context.WithCancel(context.Background())
  186. cleanupStopRunning := true
  187. cleanupCachedTLSDialer := true
  188. var cachedTLSDialer *cachedTLSDialer
  189. // Cleanup in error cases
  190. defer func() {
  191. if cleanupStopRunning {
  192. stopRunning()
  193. }
  194. if cleanupCachedTLSDialer && cachedTLSDialer != nil {
  195. cachedTLSDialer.close()
  196. }
  197. }()
  198. // Configure transport: QUIC or HTTPS or HTTP
  199. var isQUIC bool
  200. var scheme string
  201. var transport transporter
  202. var additionalHeaders http.Header
  203. var proxyUrl func(*http.Request) (*url.URL, error)
  204. if meekConfig.UseQUIC {
  205. isQUIC = true
  206. scheme = "https"
  207. udpDialer := func(ctx context.Context) (net.PacketConn, *net.UDPAddr, error) {
  208. packetConn, remoteAddr, err := NewUDPConn(
  209. ctx,
  210. meekConfig.DialAddress,
  211. dialConfig)
  212. if err != nil {
  213. return nil, nil, common.ContextError(err)
  214. }
  215. return packetConn, remoteAddr, nil
  216. }
  217. _, port, _ := net.SplitHostPort(meekConfig.DialAddress)
  218. quicDialSNIAddress := fmt.Sprintf("%s:%s", meekConfig.SNIServerName, port)
  219. transport = quic.NewQUICTransporter(
  220. ctx,
  221. udpDialer,
  222. quicDialSNIAddress,
  223. meekConfig.QUICVersion)
  224. } else if meekConfig.UseHTTPS {
  225. // Custom TLS dialer:
  226. //
  227. // 1. ignores the HTTP request address and uses the fronting domain
  228. // 2. optionally disables SNI -- SNI breaks fronting when used with certain CDNs.
  229. // 3. skips verifying the server cert.
  230. //
  231. // Reasoning for #3:
  232. //
  233. // With a TLS MiM attack in place, and server certs verified, we'll fail to connect because the client
  234. // will refuse to connect. That's not a successful outcome.
  235. //
  236. // With a MiM attack in place, and server certs not verified, we'll fail to connect if the MiM is actively
  237. // targeting Psiphon and classifying the HTTP traffic by Host header or payload signature.
  238. //
  239. // However, in the case of a passive MiM that's just recording traffic or an active MiM that's targeting
  240. // something other than Psiphon, the client will connect. This is a successful outcome.
  241. //
  242. // What is exposed to the MiM? The Host header does not contain a Psiphon server IP address, just an
  243. // unrelated, randomly generated domain name which cannot be used to block direct connections. The
  244. // Psiphon server IP is sent over meek, but it's in the encrypted cookie.
  245. //
  246. // The payload (user traffic) gets its confidentiality and integrity from the underlying SSH protocol.
  247. // So, nothing is leaked to the MiM apart from signatures which could be used to classify the traffic
  248. // as Psiphon to possibly block it; but note that not revealing that the client is Psiphon is outside
  249. // our threat model; we merely seek to evade mass blocking by taking steps that require progressively
  250. // more effort to block.
  251. //
  252. // There is a subtle attack remaining: an adversary that can MiM some CDNs but not others (and so can
  253. // classify Psiphon traffic on some CDNs but not others) may throttle non-MiM CDNs so that our server
  254. // selection always chooses tunnels to the MiM CDN (without any server cert verification, we won't
  255. // exclusively connect to non-MiM CDNs); then the adversary kills the underlying TCP connection after
  256. // some short period. This is partially mitigated by tactics mechanisms.
  257. scheme = "https"
  258. tlsConfig := &CustomTLSConfig{
  259. ClientParameters: meekConfig.ClientParameters,
  260. DialAddr: meekConfig.DialAddress,
  261. Dial: NewTCPDialer(dialConfig),
  262. SNIServerName: meekConfig.SNIServerName,
  263. SkipVerify: true,
  264. TLSProfile: meekConfig.TLSProfile,
  265. NoDefaultTLSSessionID: &meekConfig.NoDefaultTLSSessionID,
  266. RandomizedTLSProfileSeed: meekConfig.RandomizedTLSProfileSeed,
  267. TrustedCACertificatesFilename: dialConfig.TrustedCACertificatesFilename,
  268. }
  269. tlsConfig.EnableClientSessionCache()
  270. if meekConfig.UseObfuscatedSessionTickets {
  271. tlsConfig.ObfuscatedSessionTicketKey = meekConfig.MeekObfuscatedKey
  272. }
  273. tlsDialer := NewCustomTLSDialer(tlsConfig)
  274. // Pre-dial one TLS connection in order to inspect the negotiated
  275. // application protocol. Then we create an HTTP/2 or HTTP/1.1 transport
  276. // depending on which protocol was negotiated. The TLS dialer
  277. // is assumed to negotiate only "h2" or "http/1.1"; or not negotiate
  278. // an application protocol.
  279. //
  280. // We cannot rely on net/http's HTTP/2 support since it's only
  281. // activated when http.Transport.DialTLS returns a golang crypto/tls.Conn;
  282. // e.g., https://github.com/golang/go/blob/c8aec4095e089ff6ac50d18e97c3f46561f14f48/src/net/http/transport.go#L1040
  283. //
  284. // The pre-dialed connection is stored in a cachedTLSDialer, which will
  285. // return the cached pre-dialed connection to its first Dial caller, and
  286. // use the tlsDialer for all other Dials.
  287. //
  288. // cachedTLSDialer.close() must be called on all exits paths from this
  289. // function and in meek.Close() to ensure the cached conn is closed in
  290. // any case where no Dial call is made.
  291. //
  292. // The pre-dial must be interruptible so that DialMeek doesn't block and
  293. // hang/delay a shutdown or end of establishment. So the pre-dial uses
  294. // the Controller's PendingConns, not the MeekConn PendingConns. For this
  295. // purpose, a special preDialer is configured.
  296. //
  297. // Only one pre-dial attempt is made; there are no retries. This differs
  298. // from relayRoundTrip, which retries and may redial for each retry.
  299. // Retries at the pre-dial phase are less useful since there's no active
  300. // session to preserve, and establishment will simply try another server.
  301. // Note that the underlying TCPDial may still try multiple IP addreses when
  302. // the destination is a domain and it resolves to multiple IP adresses.
  303. // The pre-dial is made within the parent dial context, so that DialMeek
  304. // may be interrupted. Subsequent dials are made within the meek round trip
  305. // request context. Since http.DialTLS doesn't take a context argument
  306. // (yet; as of Go 1.9 this issue is still open: https://github.com/golang/go/issues/21526),
  307. // cachedTLSDialer is used as a conduit to send the request context.
  308. // meekConn.relayRoundTrip sets its request context into cachedTLSDialer,
  309. // and cachedTLSDialer.dial uses that context.
  310. // As DialAddr is set in the CustomTLSConfig, no address is required here.
  311. preConn, err := tlsDialer(ctx, "tcp", "")
  312. if err != nil {
  313. return nil, common.ContextError(err)
  314. }
  315. cachedTLSDialer = newCachedTLSDialer(preConn, tlsDialer)
  316. if IsTLSConnUsingHTTP2(preConn) {
  317. NoticeInfo("negotiated HTTP/2 for %s", meekConfig.DiagnosticID)
  318. transport = &http2.Transport{
  319. DialTLS: func(network, addr string, _ *tls.Config) (net.Conn, error) {
  320. return cachedTLSDialer.dial(network, addr)
  321. },
  322. }
  323. } else {
  324. transport = &http.Transport{
  325. DialTLS: func(network, addr string) (net.Conn, error) {
  326. return cachedTLSDialer.dial(network, addr)
  327. },
  328. }
  329. }
  330. } else {
  331. scheme = "http"
  332. var dialer Dialer
  333. // For HTTP, and when the meekConfig.DialAddress matches the
  334. // meekConfig.HostHeader, we let http.Transport handle proxying.
  335. // http.Transport will put the the HTTP server address in the HTTP
  336. // request line. In this one case, we can use an HTTP proxy that does
  337. // not offer CONNECT support.
  338. if strings.HasPrefix(dialConfig.UpstreamProxyURL, "http://") &&
  339. (meekConfig.DialAddress == meekConfig.HostHeader ||
  340. meekConfig.DialAddress == meekConfig.HostHeader+":80") {
  341. url, err := url.Parse(dialConfig.UpstreamProxyURL)
  342. if err != nil {
  343. return nil, common.ContextError(err)
  344. }
  345. proxyUrl = http.ProxyURL(url)
  346. // Here, the dialer must use the address that http.Transport
  347. // passes in (which will be proxy address).
  348. copyDialConfig := new(DialConfig)
  349. *copyDialConfig = *dialConfig
  350. copyDialConfig.UpstreamProxyURL = ""
  351. dialer = NewTCPDialer(copyDialConfig)
  352. } else {
  353. baseDialer := NewTCPDialer(dialConfig)
  354. // The dialer ignores any address that http.Transport will pass in
  355. // (derived from the HTTP request URL) and always dials
  356. // meekConfig.DialAddress.
  357. dialer = func(ctx context.Context, network, _ string) (net.Conn, error) {
  358. return baseDialer(ctx, network, meekConfig.DialAddress)
  359. }
  360. }
  361. httpTransport := &http.Transport{
  362. Proxy: proxyUrl,
  363. DialContext: dialer,
  364. }
  365. if proxyUrl != nil {
  366. // Wrap transport with a transport that can perform HTTP proxy auth negotiation
  367. transport, err = upstreamproxy.NewProxyAuthTransport(httpTransport, dialConfig.CustomHeaders)
  368. if err != nil {
  369. return nil, common.ContextError(err)
  370. }
  371. } else {
  372. transport = httpTransport
  373. }
  374. }
  375. url := &url.URL{
  376. Scheme: scheme,
  377. Host: meekConfig.HostHeader,
  378. Path: "/",
  379. }
  380. if meekConfig.UseHTTPS {
  381. host, _, err := net.SplitHostPort(meekConfig.DialAddress)
  382. if err != nil {
  383. return nil, common.ContextError(err)
  384. }
  385. additionalHeaders = map[string][]string{
  386. "X-Psiphon-Fronting-Address": {host},
  387. }
  388. } else {
  389. if proxyUrl == nil {
  390. additionalHeaders = dialConfig.CustomHeaders
  391. }
  392. }
  393. // The main loop of a MeekConn is run in the relay() goroutine.
  394. // A MeekConn implements net.Conn concurrency semantics:
  395. // "Multiple goroutines may invoke methods on a Conn simultaneously."
  396. //
  397. // Read() calls and relay() are synchronized by exchanging control of a single
  398. // receiveBuffer (bytes.Buffer). This single buffer may be:
  399. // - in the emptyReceiveBuffer channel when it is available and empty;
  400. // - in the partialReadBuffer channel when it is available and contains data;
  401. // - in the fullReadBuffer channel when it is available and full of data;
  402. // - "checked out" by relay or Read when they are are writing to or reading from the
  403. // buffer, respectively.
  404. // relay() will obtain the buffer from either the empty or partial channel but block when
  405. // the buffer is full. Read will obtain the buffer from the partial or full channel when
  406. // there is data to read but block when the buffer is empty.
  407. // Write() calls and relay() are synchronized in a similar way, using a single
  408. // sendBuffer.
  409. meek = &MeekConn{
  410. clientParameters: meekConfig.ClientParameters,
  411. networkLatencyMultiplier: meekConfig.NetworkLatencyMultiplier,
  412. isQUIC: isQUIC,
  413. url: url,
  414. additionalHeaders: additionalHeaders,
  415. cachedTLSDialer: cachedTLSDialer,
  416. transport: transport,
  417. isClosed: false,
  418. runCtx: runCtx,
  419. stopRunning: stopRunning,
  420. relayWaitGroup: new(sync.WaitGroup),
  421. roundTripperOnly: meekConfig.RoundTripperOnly,
  422. }
  423. // stopRunning and cachedTLSDialer will now be closed in meek.Close()
  424. cleanupStopRunning = false
  425. cleanupCachedTLSDialer = false
  426. // Allocate relay resources, including buffers and running the relay
  427. // go routine, only when running in relay mode.
  428. if !meek.roundTripperOnly {
  429. cookie, limitRequestPayloadLength, redialTLSProbability, err :=
  430. makeMeekObfuscationValues(
  431. meek.getCustomClientParameters(),
  432. meekConfig.MeekCookieEncryptionPublicKey,
  433. meekConfig.MeekObfuscatedKey,
  434. meekConfig.MeekObfuscatorPaddingSeed,
  435. meekConfig.ClientTunnelProtocol,
  436. "")
  437. if err != nil {
  438. return nil, common.ContextError(err)
  439. }
  440. meek.cookie = cookie
  441. meek.cookieSize = len(cookie.Name) + len(cookie.Value)
  442. meek.limitRequestPayloadLength = limitRequestPayloadLength
  443. meek.redialTLSProbability = redialTLSProbability
  444. p := meek.getCustomClientParameters()
  445. if p.Bool(parameters.MeekLimitBufferSizes) {
  446. meek.fullReceiveBufferLength = p.Int(parameters.MeekLimitedFullReceiveBufferLength)
  447. meek.readPayloadChunkLength = p.Int(parameters.MeekLimitedReadPayloadChunkLength)
  448. } else {
  449. meek.fullReceiveBufferLength = p.Int(parameters.MeekFullReceiveBufferLength)
  450. meek.readPayloadChunkLength = p.Int(parameters.MeekReadPayloadChunkLength)
  451. }
  452. meek.emptyReceiveBuffer = make(chan *bytes.Buffer, 1)
  453. meek.partialReceiveBuffer = make(chan *bytes.Buffer, 1)
  454. meek.fullReceiveBuffer = make(chan *bytes.Buffer, 1)
  455. meek.emptySendBuffer = make(chan *bytes.Buffer, 1)
  456. meek.partialSendBuffer = make(chan *bytes.Buffer, 1)
  457. meek.fullSendBuffer = make(chan *bytes.Buffer, 1)
  458. meek.emptyReceiveBuffer <- new(bytes.Buffer)
  459. meek.emptySendBuffer <- new(bytes.Buffer)
  460. meek.relayWaitGroup.Add(1)
  461. go meek.relay()
  462. } else {
  463. meek.meekCookieEncryptionPublicKey = meekConfig.MeekCookieEncryptionPublicKey
  464. meek.meekObfuscatedKey = meekConfig.MeekObfuscatedKey
  465. meek.meekObfuscatorPaddingSeed = meekConfig.MeekObfuscatorPaddingSeed
  466. meek.clientTunnelProtocol = meekConfig.ClientTunnelProtocol
  467. }
  468. return meek, nil
  469. }
  470. type cachedTLSDialer struct {
  471. usedCachedConn int32
  472. cachedConn net.Conn
  473. dialer Dialer
  474. mutex sync.Mutex
  475. requestCtx context.Context
  476. }
  477. func newCachedTLSDialer(cachedConn net.Conn, dialer Dialer) *cachedTLSDialer {
  478. return &cachedTLSDialer{
  479. cachedConn: cachedConn,
  480. dialer: dialer,
  481. }
  482. }
  483. func (c *cachedTLSDialer) setRequestContext(requestCtx context.Context) {
  484. // Note: not using sync.Value since underlying type of requestCtx may change.
  485. c.mutex.Lock()
  486. defer c.mutex.Unlock()
  487. c.requestCtx = requestCtx
  488. }
  489. func (c *cachedTLSDialer) dial(network, addr string) (net.Conn, error) {
  490. if atomic.CompareAndSwapInt32(&c.usedCachedConn, 0, 1) {
  491. conn := c.cachedConn
  492. c.cachedConn = nil
  493. return conn, nil
  494. }
  495. c.mutex.Lock()
  496. ctx := c.requestCtx
  497. c.mutex.Unlock()
  498. if ctx == nil {
  499. ctx = context.Background()
  500. }
  501. return c.dialer(ctx, network, addr)
  502. }
  503. func (c *cachedTLSDialer) close() {
  504. if atomic.CompareAndSwapInt32(&c.usedCachedConn, 0, 1) {
  505. c.cachedConn.Close()
  506. c.cachedConn = nil
  507. }
  508. }
  509. // Close terminates the meek connection. Close waits for the relay goroutine
  510. // to stop (in relay mode) and releases HTTP transport resources.
  511. // A mutex is required to support net.Conn concurrency semantics.
  512. func (meek *MeekConn) Close() (err error) {
  513. meek.mutex.Lock()
  514. isClosed := meek.isClosed
  515. meek.isClosed = true
  516. meek.mutex.Unlock()
  517. if !isClosed {
  518. meek.stopRunning()
  519. if meek.cachedTLSDialer != nil {
  520. meek.cachedTLSDialer.close()
  521. }
  522. // stopRunning interrupts HTTP requests in progress by closing the context
  523. // associated with the request. In the case of h2quic.RoundTripper, testing
  524. // indicates that quic-go.receiveStream.readImpl in _not_ interrupted in
  525. // this case, and so an in-flight FRONTED-MEEK-QUIC round trip may hang shutdown
  526. // in relayRoundTrip->readPayload->...->quic-go.receiveStream.readImpl.
  527. //
  528. // To workaround this, we call CloseIdleConnections _before_ Wait, as, in
  529. // the case of QUICTransporter, this closes the underlying UDP sockets which
  530. // interrupts any blocking I/O calls.
  531. //
  532. // The standard CloseIdleConnections call _after_ wait is for the net/http
  533. // case: it only closes idle connections, so the call should be after wait.
  534. // This call is intended to clean up all network resources deterministically
  535. // before Close returns.
  536. if meek.isQUIC {
  537. meek.transport.CloseIdleConnections()
  538. }
  539. meek.relayWaitGroup.Wait()
  540. meek.transport.CloseIdleConnections()
  541. }
  542. return nil
  543. }
  544. // IsClosed implements the Closer interface. The return value
  545. // indicates whether the MeekConn has been closed.
  546. func (meek *MeekConn) IsClosed() bool {
  547. meek.mutex.Lock()
  548. isClosed := meek.isClosed
  549. meek.mutex.Unlock()
  550. return isClosed
  551. }
  552. // GetMetrics implements the common.MetricsSource interface.
  553. func (meek *MeekConn) GetMetrics() common.LogFields {
  554. logFields := make(common.LogFields)
  555. logFields["meek_cookie_size"] = meek.cookieSize
  556. logFields["meek_limit_request"] = meek.limitRequestPayloadLength
  557. return logFields
  558. }
  559. // RoundTrip makes a request to the meek server and returns the response.
  560. // A new, obfuscated meek cookie is created for every request. The specified
  561. // end point is recorded in the cookie and is not exposed as plaintext in the
  562. // meek traffic. The caller is responsible for obfuscating the request body.
  563. //
  564. // RoundTrip is not safe for concurrent use, and Close must not be called
  565. // concurrently. The caller must ensure onlt one RoundTrip call is active
  566. // at once and that it completes before calling Close.
  567. //
  568. // RoundTrip is only available in round tripper mode.
  569. func (meek *MeekConn) RoundTrip(
  570. ctx context.Context, endPoint string, requestBody []byte) ([]byte, error) {
  571. if !meek.roundTripperOnly {
  572. return nil, common.ContextError(errors.New("operation unsupported"))
  573. }
  574. cookie, _, _, err := makeMeekObfuscationValues(
  575. meek.getCustomClientParameters(),
  576. meek.meekCookieEncryptionPublicKey,
  577. meek.meekObfuscatedKey,
  578. meek.meekObfuscatorPaddingSeed,
  579. meek.clientTunnelProtocol,
  580. endPoint)
  581. if err != nil {
  582. return nil, common.ContextError(err)
  583. }
  584. // Note:
  585. //
  586. // - multiple, concurrent RoundTrip calls are unsafe due to the
  587. // setRequestContext calls in newRequest.
  588. //
  589. // - concurrent Close and RoundTrip calls are unsafe as Close
  590. // does not synchronize with RoundTrip before calling
  591. // meek.transport.CloseIdleConnections(), so resources could
  592. // be left open.
  593. //
  594. // At this time, RoundTrip is used for tactics in Controller and
  595. // the concurrency constraints are satisfied.
  596. request, cancelFunc, err := meek.newRequest(
  597. ctx, cookie, bytes.NewReader(requestBody), 0)
  598. if err != nil {
  599. return nil, common.ContextError(err)
  600. }
  601. defer cancelFunc()
  602. // Workaround for h2quic.RoundTripper context issue. See comment in
  603. // MeekConn.Close.
  604. if meek.isQUIC {
  605. go func() {
  606. <-request.Context().Done()
  607. meek.transport.CloseIdleConnections()
  608. }()
  609. }
  610. response, err := meek.transport.RoundTrip(request)
  611. if err == nil {
  612. defer response.Body.Close()
  613. if response.StatusCode != http.StatusOK {
  614. err = fmt.Errorf("unexpected response status code: %d", response.StatusCode)
  615. }
  616. }
  617. if err != nil {
  618. return nil, common.ContextError(err)
  619. }
  620. responseBody, err := ioutil.ReadAll(response.Body)
  621. if err != nil {
  622. return nil, common.ContextError(err)
  623. }
  624. return responseBody, nil
  625. }
  626. // Read reads data from the connection.
  627. // net.Conn Deadlines are ignored. net.Conn concurrency semantics are supported.
  628. func (meek *MeekConn) Read(buffer []byte) (n int, err error) {
  629. if meek.roundTripperOnly {
  630. return 0, common.ContextError(errors.New("operation unsupported"))
  631. }
  632. if meek.IsClosed() {
  633. return 0, common.ContextError(errors.New("meek connection is closed"))
  634. }
  635. // Block until there is received data to consume
  636. var receiveBuffer *bytes.Buffer
  637. select {
  638. case receiveBuffer = <-meek.partialReceiveBuffer:
  639. case receiveBuffer = <-meek.fullReceiveBuffer:
  640. case <-meek.runCtx.Done():
  641. return 0, common.ContextError(errors.New("meek connection has closed"))
  642. }
  643. n, err = receiveBuffer.Read(buffer)
  644. meek.replaceReceiveBuffer(receiveBuffer)
  645. return n, err
  646. }
  647. // Write writes data to the connection.
  648. // net.Conn Deadlines are ignored. net.Conn concurrency semantics are supported.
  649. func (meek *MeekConn) Write(buffer []byte) (n int, err error) {
  650. if meek.roundTripperOnly {
  651. return 0, common.ContextError(errors.New("operation unsupported"))
  652. }
  653. if meek.IsClosed() {
  654. return 0, common.ContextError(errors.New("meek connection is closed"))
  655. }
  656. // Repeats until all n bytes are written
  657. n = len(buffer)
  658. for len(buffer) > 0 {
  659. // Block until there is capacity in the send buffer
  660. var sendBuffer *bytes.Buffer
  661. select {
  662. case sendBuffer = <-meek.emptySendBuffer:
  663. case sendBuffer = <-meek.partialSendBuffer:
  664. case <-meek.runCtx.Done():
  665. return 0, common.ContextError(errors.New("meek connection has closed"))
  666. }
  667. writeLen := meek.limitRequestPayloadLength - sendBuffer.Len()
  668. if writeLen > 0 {
  669. if writeLen > len(buffer) {
  670. writeLen = len(buffer)
  671. }
  672. _, err = sendBuffer.Write(buffer[:writeLen])
  673. buffer = buffer[writeLen:]
  674. }
  675. meek.replaceSendBuffer(sendBuffer)
  676. }
  677. return n, err
  678. }
  679. // LocalAddr is a stub implementation of net.Conn.LocalAddr
  680. func (meek *MeekConn) LocalAddr() net.Addr {
  681. return nil
  682. }
  683. // RemoteAddr is a stub implementation of net.Conn.RemoteAddr
  684. func (meek *MeekConn) RemoteAddr() net.Addr {
  685. return nil
  686. }
  687. // SetDeadline is a stub implementation of net.Conn.SetDeadline
  688. func (meek *MeekConn) SetDeadline(t time.Time) error {
  689. return common.ContextError(errors.New("not supported"))
  690. }
  691. // SetReadDeadline is a stub implementation of net.Conn.SetReadDeadline
  692. func (meek *MeekConn) SetReadDeadline(t time.Time) error {
  693. return common.ContextError(errors.New("not supported"))
  694. }
  695. // SetWriteDeadline is a stub implementation of net.Conn.SetWriteDeadline
  696. func (meek *MeekConn) SetWriteDeadline(t time.Time) error {
  697. return common.ContextError(errors.New("not supported"))
  698. }
  699. func (meek *MeekConn) replaceReceiveBuffer(receiveBuffer *bytes.Buffer) {
  700. switch {
  701. case receiveBuffer.Len() == 0:
  702. meek.emptyReceiveBuffer <- receiveBuffer
  703. case receiveBuffer.Len() >= meek.fullReceiveBufferLength:
  704. meek.fullReceiveBuffer <- receiveBuffer
  705. default:
  706. meek.partialReceiveBuffer <- receiveBuffer
  707. }
  708. }
  709. func (meek *MeekConn) replaceSendBuffer(sendBuffer *bytes.Buffer) {
  710. switch {
  711. case sendBuffer.Len() == 0:
  712. meek.emptySendBuffer <- sendBuffer
  713. case sendBuffer.Len() >= meek.limitRequestPayloadLength:
  714. meek.fullSendBuffer <- sendBuffer
  715. default:
  716. meek.partialSendBuffer <- sendBuffer
  717. }
  718. }
  719. // relay sends and receives tunneled traffic (payload). An HTTP request is
  720. // triggered when data is in the write queue or at a polling interval.
  721. // There's a geometric increase, up to a maximum, in the polling interval when
  722. // no data is exchanged. Only one HTTP request is in flight at a time.
  723. func (meek *MeekConn) relay() {
  724. // Note: meek.Close() calls here in relay() are made asynchronously
  725. // (using goroutines) since Close() will wait on this WaitGroup.
  726. defer meek.relayWaitGroup.Done()
  727. p := meek.getCustomClientParameters()
  728. interval := prng.JitterDuration(
  729. p.Duration(parameters.MeekMinPollInterval),
  730. p.Float(parameters.MeekMinPollIntervalJitter))
  731. p.Close()
  732. timeout := time.NewTimer(interval)
  733. defer timeout.Stop()
  734. for {
  735. timeout.Reset(interval)
  736. // Block until there is payload to send or it is time to poll
  737. var sendBuffer *bytes.Buffer
  738. select {
  739. case sendBuffer = <-meek.partialSendBuffer:
  740. case sendBuffer = <-meek.fullSendBuffer:
  741. case <-timeout.C:
  742. // In the polling case, send an empty payload
  743. case <-meek.runCtx.Done():
  744. // Drop through to second Done() check
  745. }
  746. // Check Done() again, to ensure it takes precedence
  747. select {
  748. case <-meek.runCtx.Done():
  749. return
  750. default:
  751. }
  752. sendPayloadSize := 0
  753. if sendBuffer != nil {
  754. sendPayloadSize = sendBuffer.Len()
  755. }
  756. // relayRoundTrip will replace sendBuffer (by calling replaceSendBuffer). This
  757. // is a compromise to conserve memory. Using a second buffer here, we could
  758. // copy sendBuffer and immediately replace it, unblocking meekConn.Write() and
  759. // allowing more upstream payload to immediately enqueue. Instead, the request
  760. // payload is read directly from sendBuffer, including retries. Only once the
  761. // server has acknowledged the request payload is sendBuffer replaced. This
  762. // still allows meekConn.Write() to unblock before the round trip response is
  763. // read.
  764. receivedPayloadSize, err := meek.relayRoundTrip(sendBuffer)
  765. if err != nil {
  766. select {
  767. case <-meek.runCtx.Done():
  768. // In this case, meek.relayRoundTrip encountered Done(). Exit without
  769. // logging error.
  770. return
  771. default:
  772. }
  773. NoticeAlert("%s", common.ContextError(err))
  774. go meek.Close()
  775. return
  776. }
  777. // Periodically re-dial the underlying TLS connection.
  778. if prng.FlipWeightedCoin(meek.redialTLSProbability) {
  779. meek.transport.CloseIdleConnections()
  780. }
  781. // Calculate polling interval. When data is received,
  782. // immediately request more. Otherwise, schedule next
  783. // poll with exponential back off. Jitter and coin
  784. // flips are used to avoid trivial, static traffic
  785. // timing patterns.
  786. p := meek.getCustomClientParameters()
  787. if receivedPayloadSize > 0 || sendPayloadSize > 0 {
  788. interval = 0
  789. } else if interval == 0 {
  790. interval = prng.JitterDuration(
  791. p.Duration(parameters.MeekMinPollInterval),
  792. p.Float(parameters.MeekMinPollIntervalJitter))
  793. } else {
  794. if p.WeightedCoinFlip(parameters.MeekApplyPollIntervalMultiplierProbability) {
  795. interval =
  796. time.Duration(float64(interval) *
  797. p.Float(parameters.MeekPollIntervalMultiplier))
  798. }
  799. interval = prng.JitterDuration(
  800. interval,
  801. p.Float(parameters.MeekPollIntervalJitter))
  802. if interval >= p.Duration(parameters.MeekMaxPollInterval) {
  803. interval = prng.JitterDuration(
  804. p.Duration(parameters.MeekMaxPollInterval),
  805. p.Float(parameters.MeekMaxPollIntervalJitter))
  806. }
  807. }
  808. p.Close()
  809. }
  810. }
  811. // readCloseSignaller is an io.ReadCloser wrapper for an io.Reader
  812. // that is passed, as the request body, to http.Transport.RoundTrip.
  813. // readCloseSignaller adds the AwaitClosed call, which is used
  814. // to schedule recycling the buffer underlying the reader only after
  815. // RoundTrip has called Close and will no longer use the buffer.
  816. // See: https://golang.org/pkg/net/http/#RoundTripper
  817. type readCloseSignaller struct {
  818. context context.Context
  819. reader io.Reader
  820. closed chan struct{}
  821. }
  822. func NewReadCloseSignaller(
  823. context context.Context,
  824. reader io.Reader) *readCloseSignaller {
  825. return &readCloseSignaller{
  826. context: context,
  827. reader: reader,
  828. closed: make(chan struct{}, 1),
  829. }
  830. }
  831. func (r *readCloseSignaller) Read(p []byte) (int, error) {
  832. return r.reader.Read(p)
  833. }
  834. func (r *readCloseSignaller) Close() error {
  835. select {
  836. case r.closed <- *new(struct{}):
  837. default:
  838. }
  839. return nil
  840. }
  841. func (r *readCloseSignaller) AwaitClosed() bool {
  842. select {
  843. case <-r.context.Done():
  844. case <-r.closed:
  845. return true
  846. }
  847. return false
  848. }
  849. // newRequest performs common request setup for both relay and round
  850. // tripper modes.
  851. //
  852. // newRequest is not safe for concurrent calls due to its use of
  853. // setRequestContext.
  854. //
  855. // The caller must call the returned cancelFunc.
  856. func (meek *MeekConn) newRequest(
  857. ctx context.Context,
  858. cookie *http.Cookie,
  859. body io.Reader,
  860. contentLength int) (*http.Request, context.CancelFunc, error) {
  861. var requestCtx context.Context
  862. var cancelFunc context.CancelFunc
  863. if ctx != nil {
  864. requestCtx, cancelFunc = context.WithCancel(ctx)
  865. } else {
  866. // - meek.stopRunning() will abort a round trip in flight
  867. // - round trip will abort if it exceeds timeout
  868. requestCtx, cancelFunc = context.WithTimeout(
  869. meek.runCtx,
  870. meek.getCustomClientParameters().Duration(parameters.MeekRoundTripTimeout))
  871. }
  872. // Ensure dials are made within the current request context.
  873. if meek.isQUIC {
  874. meek.transport.(*quic.QUICTransporter).SetRequestContext(requestCtx)
  875. } else if meek.cachedTLSDialer != nil {
  876. meek.cachedTLSDialer.setRequestContext(requestCtx)
  877. }
  878. request, err := http.NewRequest("POST", meek.url.String(), body)
  879. if err != nil {
  880. cancelFunc()
  881. return nil, nil, common.ContextError(err)
  882. }
  883. request = request.WithContext(requestCtx)
  884. // Content-Length may not be be set automatically due to the
  885. // underlying type of requestBody.
  886. if contentLength > 0 {
  887. request.ContentLength = int64(contentLength)
  888. }
  889. meek.addAdditionalHeaders(request)
  890. request.Header.Set("Content-Type", "application/octet-stream")
  891. if cookie == nil {
  892. cookie = meek.cookie
  893. }
  894. request.AddCookie(cookie)
  895. return request, cancelFunc, nil
  896. }
  897. // relayRoundTrip configures and makes the actual HTTP POST request
  898. func (meek *MeekConn) relayRoundTrip(sendBuffer *bytes.Buffer) (int64, error) {
  899. // Retries are made when the round trip fails. This adds resiliency
  900. // to connection interruption and intermittent failures.
  901. //
  902. // At least one retry is always attempted, and retries continue
  903. // while still within a brief deadline -- 5 seconds, currently the
  904. // deadline for an actively probed SSH connection to timeout. There
  905. // is a brief delay between retries, allowing for intermittent
  906. // failure states to resolve.
  907. //
  908. // Failure may occur at various stages of the HTTP request:
  909. //
  910. // 1. Before the request begins. In this case, the entire request
  911. // may be rerun.
  912. //
  913. // 2. While sending the request payload. In this case, the client
  914. // must resend its request payload. The server will not have
  915. // relayed its partially received request payload.
  916. //
  917. // 3. After sending the request payload but before receiving
  918. // a response. The client cannot distinguish between case 2 and
  919. // this case, case 3. The client resends its payload and the
  920. // server detects this and skips relaying the request payload.
  921. //
  922. // 4. While reading the response payload. The client will omit its
  923. // request payload when retrying, as the server has already
  924. // acknowledged it. The client will also indicate to the server
  925. // the amount of response payload already received, and the
  926. // server will skip resending the indicated amount of response
  927. // payload.
  928. //
  929. // Retries are indicated to the server by adding a Range header,
  930. // which includes the response payload resend position.
  931. defer func() {
  932. // Ensure sendBuffer is replaced, even in error code paths.
  933. if sendBuffer != nil {
  934. sendBuffer.Truncate(0)
  935. meek.replaceSendBuffer(sendBuffer)
  936. }
  937. }()
  938. retries := uint(0)
  939. p := meek.getCustomClientParameters()
  940. retryDeadline := monotime.Now().Add(p.Duration(parameters.MeekRoundTripRetryDeadline))
  941. retryDelay := p.Duration(parameters.MeekRoundTripRetryMinDelay)
  942. retryMaxDelay := p.Duration(parameters.MeekRoundTripRetryMaxDelay)
  943. retryMultiplier := p.Float(parameters.MeekRoundTripRetryMultiplier)
  944. p.Close()
  945. serverAcknowledgedRequestPayload := false
  946. receivedPayloadSize := int64(0)
  947. for try := 0; ; try++ {
  948. // Omit the request payload when retrying after receiving a
  949. // partial server response.
  950. var signaller *readCloseSignaller
  951. var requestBody io.ReadCloser
  952. contentLength := 0
  953. if !serverAcknowledgedRequestPayload && sendBuffer != nil {
  954. // sendBuffer will be replaced once the data is no longer needed,
  955. // when RoundTrip calls Close on the Body; this allows meekConn.Write()
  956. // to unblock and start buffering data for the next roung trip while
  957. // still reading the current round trip response. signaller provides
  958. // the hook for awaiting RoundTrip's call to Close.
  959. signaller = NewReadCloseSignaller(meek.runCtx, bytes.NewReader(sendBuffer.Bytes()))
  960. requestBody = signaller
  961. contentLength = sendBuffer.Len()
  962. }
  963. request, cancelFunc, err := meek.newRequest(
  964. nil,
  965. nil,
  966. requestBody,
  967. contentLength)
  968. if err != nil {
  969. // Don't retry when can't initialize a Request
  970. return 0, common.ContextError(err)
  971. }
  972. expectedStatusCode := http.StatusOK
  973. // When retrying, add a Range header to indicate how much
  974. // of the response was already received.
  975. if try > 0 {
  976. expectedStatusCode = http.StatusPartialContent
  977. request.Header.Set("Range", fmt.Sprintf("bytes=%d-", receivedPayloadSize))
  978. }
  979. response, err := meek.transport.RoundTrip(request)
  980. // Wait for RoundTrip to call Close on the request body, when
  981. // there is one. This is necessary to ensure it's safe to
  982. // subsequently replace sendBuffer in both the success and
  983. // error cases.
  984. if signaller != nil {
  985. if !signaller.AwaitClosed() {
  986. // AwaitClosed encountered Done(). Abort immediately. Do not
  987. // replace sendBuffer, as we cannot be certain RoundTrip is
  988. // done with it. MeekConn.Write will exit on Done and not hang
  989. // awaiting sendBuffer.
  990. sendBuffer = nil
  991. return 0, common.ContextError(errors.New("meek connection has closed"))
  992. }
  993. }
  994. if err != nil {
  995. select {
  996. case <-meek.runCtx.Done():
  997. // Exit without retrying and without logging error.
  998. return 0, common.ContextError(err)
  999. default:
  1000. }
  1001. NoticeAlert("meek round trip failed: %s", err)
  1002. // ...continue to retry
  1003. }
  1004. if err == nil {
  1005. if response.StatusCode != expectedStatusCode &&
  1006. // Certain http servers return 200 OK where we expect 206, so accept that.
  1007. !(expectedStatusCode == http.StatusPartialContent && response.StatusCode == http.StatusOK) {
  1008. // Don't retry when the status code is incorrect
  1009. response.Body.Close()
  1010. return 0, common.ContextError(
  1011. fmt.Errorf(
  1012. "unexpected status code: %d instead of %d",
  1013. response.StatusCode, expectedStatusCode))
  1014. }
  1015. // Update meek session cookie
  1016. for _, c := range response.Cookies() {
  1017. if meek.cookie.Name == c.Name {
  1018. meek.cookie.Value = c.Value
  1019. break
  1020. }
  1021. }
  1022. // Received the response status code, so the server
  1023. // must have received the request payload.
  1024. serverAcknowledgedRequestPayload = true
  1025. // sendBuffer is now no longer required for retries, and the
  1026. // buffer may be replaced; this allows meekConn.Write() to unblock
  1027. // and start buffering data for the next round trip while still
  1028. // reading the current round trip response.
  1029. if sendBuffer != nil {
  1030. // Assumes signaller.AwaitClosed is called above, so
  1031. // sendBuffer will no longer be accessed by RoundTrip.
  1032. sendBuffer.Truncate(0)
  1033. meek.replaceSendBuffer(sendBuffer)
  1034. sendBuffer = nil
  1035. }
  1036. readPayloadSize, err := meek.readPayload(response.Body)
  1037. response.Body.Close()
  1038. // receivedPayloadSize is the number of response
  1039. // payload bytes received and relayed. A retry can
  1040. // resume after this position.
  1041. receivedPayloadSize += readPayloadSize
  1042. if err != nil {
  1043. NoticeAlert("meek read payload failed: %s", err)
  1044. // ...continue to retry
  1045. } else {
  1046. // Round trip completed successfully
  1047. break
  1048. }
  1049. }
  1050. // Release context resources now.
  1051. cancelFunc()
  1052. // Either the request failed entirely, or there was a failure
  1053. // streaming the response payload. Always retry once. Then
  1054. // retry if time remains; when the next delay exceeds the time
  1055. // remaining until the deadline, do not retry.
  1056. now := monotime.Now()
  1057. if retries >= 1 &&
  1058. (now.After(retryDeadline) || retryDeadline.Sub(now) <= retryDelay) {
  1059. return 0, common.ContextError(err)
  1060. }
  1061. retries += 1
  1062. delayTimer := time.NewTimer(retryDelay)
  1063. select {
  1064. case <-delayTimer.C:
  1065. case <-meek.runCtx.Done():
  1066. delayTimer.Stop()
  1067. return 0, common.ContextError(err)
  1068. }
  1069. // Increase the next delay, to back off and avoid excessive
  1070. // activity in conditions such as no network connectivity.
  1071. retryDelay = time.Duration(
  1072. float64(retryDelay) * retryMultiplier)
  1073. if retryDelay >= retryMaxDelay {
  1074. retryDelay = retryMaxDelay
  1075. }
  1076. }
  1077. return receivedPayloadSize, nil
  1078. }
  1079. // Add additional headers to the HTTP request using the same method we use for adding
  1080. // custom headers to HTTP proxy requests.
  1081. func (meek *MeekConn) addAdditionalHeaders(request *http.Request) {
  1082. for name, value := range meek.additionalHeaders {
  1083. // hack around special case of "Host" header
  1084. // https://golang.org/src/net/http/request.go#L474
  1085. // using URL.Opaque, see URL.RequestURI() https://golang.org/src/net/url/url.go#L915
  1086. if name == "Host" {
  1087. if len(value) > 0 {
  1088. if request.URL.Opaque == "" {
  1089. request.URL.Opaque = request.URL.Scheme + "://" + request.Host + request.URL.RequestURI()
  1090. }
  1091. request.Host = value[0]
  1092. }
  1093. } else {
  1094. request.Header[name] = value
  1095. }
  1096. }
  1097. }
  1098. // readPayload reads the HTTP response in chunks, making the read buffer available
  1099. // to MeekConn.Read() calls after each chunk; the intention is to allow bytes to
  1100. // flow back to the reader as soon as possible instead of buffering the entire payload.
  1101. //
  1102. // When readPayload returns an error, the totalSize output is remains valid -- it's the
  1103. // number of payload bytes successfully read and relayed.
  1104. func (meek *MeekConn) readPayload(
  1105. receivedPayload io.ReadCloser) (totalSize int64, err error) {
  1106. defer receivedPayload.Close()
  1107. totalSize = 0
  1108. for {
  1109. reader := io.LimitReader(receivedPayload, int64(meek.readPayloadChunkLength))
  1110. // Block until there is capacity in the receive buffer
  1111. var receiveBuffer *bytes.Buffer
  1112. select {
  1113. case receiveBuffer = <-meek.emptyReceiveBuffer:
  1114. case receiveBuffer = <-meek.partialReceiveBuffer:
  1115. case <-meek.runCtx.Done():
  1116. return 0, nil
  1117. }
  1118. // Note: receiveBuffer size may exceed meek.fullReceiveBufferLength by up to the size
  1119. // of one received payload. The meek.fullReceiveBufferLength value is just a guideline.
  1120. n, err := receiveBuffer.ReadFrom(reader)
  1121. meek.replaceReceiveBuffer(receiveBuffer)
  1122. totalSize += n
  1123. if err != nil {
  1124. return totalSize, common.ContextError(err)
  1125. }
  1126. if n == 0 {
  1127. break
  1128. }
  1129. }
  1130. return totalSize, nil
  1131. }
  1132. // makeMeekObfuscationValues creates the meek cookie, to be sent with initial
  1133. // meek HTTP request, and other meek obfuscation values. The cookies contains
  1134. // obfuscated metadata, including meek version and other protocol information.
  1135. //
  1136. // In round tripper mode, the cookie contains the destination endpoint for the
  1137. // round trip request.
  1138. //
  1139. // In relay mode, the server will create a session using the cookie values and
  1140. // send the session ID back to the client via Set-Cookie header. The client
  1141. // must use that value with all consequent HTTP requests.
  1142. //
  1143. // In plain HTTP meek protocols, the cookie is visible over the adversary
  1144. // network, so the cookie is encrypted and obfuscated.
  1145. //
  1146. // Obsolete meek cookie fields used by the legacy server stack are no longer
  1147. // sent. These include ServerAddress and SessionID.
  1148. //
  1149. // The request payload limit and TLS redial probability apply only to relay
  1150. // mode and are selected once and used for the duration of a meek connction.
  1151. func makeMeekObfuscationValues(
  1152. p parameters.ClientParametersAccessor,
  1153. meekCookieEncryptionPublicKey string,
  1154. meekObfuscatedKey string,
  1155. meekObfuscatorPaddingPRNGSeed *prng.Seed,
  1156. clientTunnelProtocol string,
  1157. endPoint string,
  1158. ) (cookie *http.Cookie,
  1159. limitRequestPayloadLength int,
  1160. redialTLSProbability float64,
  1161. err error) {
  1162. cookieData := &protocol.MeekCookieData{
  1163. MeekProtocolVersion: MEEK_PROTOCOL_VERSION,
  1164. ClientTunnelProtocol: clientTunnelProtocol,
  1165. EndPoint: endPoint,
  1166. }
  1167. serializedCookie, err := json.Marshal(cookieData)
  1168. if err != nil {
  1169. return nil, 0, 0, common.ContextError(err)
  1170. }
  1171. // Encrypt the JSON data
  1172. // NaCl box is used for encryption. The peer public key comes from the server entry.
  1173. // Nonce is always all zeros, and is not sent in the cookie (the server also uses an all-zero nonce).
  1174. // http://nacl.cace-project.eu/box.html:
  1175. // "There is no harm in having the same nonce for different messages if the {sender, receiver} sets are
  1176. // different. This is true even if the sets overlap. For example, a sender can use the same nonce for two
  1177. // different messages if the messages are sent to two different public keys."
  1178. var nonce [24]byte
  1179. var publicKey [32]byte
  1180. decodedPublicKey, err := base64.StdEncoding.DecodeString(meekCookieEncryptionPublicKey)
  1181. if err != nil {
  1182. return nil, 0, 0, common.ContextError(err)
  1183. }
  1184. copy(publicKey[:], decodedPublicKey)
  1185. ephemeralPublicKey, ephemeralPrivateKey, err := box.GenerateKey(rand.Reader)
  1186. if err != nil {
  1187. return nil, 0, 0, common.ContextError(err)
  1188. }
  1189. box := box.Seal(nil, serializedCookie, &nonce, &publicKey, ephemeralPrivateKey)
  1190. encryptedCookie := make([]byte, 32+len(box))
  1191. copy(encryptedCookie[0:32], ephemeralPublicKey[0:32])
  1192. copy(encryptedCookie[32:], box)
  1193. maxPadding := p.Int(parameters.MeekCookieMaxPadding)
  1194. // Obfuscate the encrypted data
  1195. obfuscator, err := obfuscator.NewClientObfuscator(
  1196. &obfuscator.ObfuscatorConfig{
  1197. Keyword: meekObfuscatedKey,
  1198. PaddingPRNGSeed: meekObfuscatorPaddingPRNGSeed,
  1199. MaxPadding: &maxPadding})
  1200. if err != nil {
  1201. return nil, 0, 0, common.ContextError(err)
  1202. }
  1203. obfuscatedCookie := obfuscator.SendSeedMessage()
  1204. seedLen := len(obfuscatedCookie)
  1205. obfuscatedCookie = append(obfuscatedCookie, encryptedCookie...)
  1206. obfuscator.ObfuscateClientToServer(obfuscatedCookie[seedLen:])
  1207. cookieNamePRNG, err := obfuscator.GetDerivedPRNG("meek-cookie-name")
  1208. if err != nil {
  1209. return nil, 0, 0, common.ContextError(err)
  1210. }
  1211. // Format the HTTP cookie
  1212. // The format is <random letter 'A'-'Z'>=<base64 data>, which is intended to match common cookie formats.
  1213. A := int('A')
  1214. Z := int('Z')
  1215. // letterIndex is integer in range [int('A'), int('Z')]
  1216. letterIndex := cookieNamePRNG.Intn(Z - A + 1)
  1217. cookie = &http.Cookie{
  1218. Name: string(byte(A + letterIndex)),
  1219. Value: base64.StdEncoding.EncodeToString(obfuscatedCookie)}
  1220. limitRequestPayloadLength = MEEK_MAX_REQUEST_PAYLOAD_LENGTH
  1221. redialTLSProbability = 0.0
  1222. tunnelProtocols := p.TunnelProtocols(parameters.MeekTrafficShapingLimitProtocols)
  1223. if (len(tunnelProtocols) == 0 ||
  1224. common.Contains(tunnelProtocols, clientTunnelProtocol)) &&
  1225. p.WeightedCoinFlip(parameters.MeekTrafficShapingProbability) {
  1226. limitRequestPayloadLengthPRNG, err := obfuscator.GetDerivedPRNG(
  1227. "meek-limit-request-payload-length")
  1228. if err != nil {
  1229. return nil, 0, 0, common.ContextError(err)
  1230. }
  1231. minLength := p.Int(parameters.MeekMinLimitRequestPayloadLength)
  1232. if minLength > MEEK_MAX_REQUEST_PAYLOAD_LENGTH {
  1233. minLength = MEEK_MAX_REQUEST_PAYLOAD_LENGTH
  1234. }
  1235. maxLength := p.Int(parameters.MeekMaxLimitRequestPayloadLength)
  1236. if maxLength > MEEK_MAX_REQUEST_PAYLOAD_LENGTH {
  1237. maxLength = MEEK_MAX_REQUEST_PAYLOAD_LENGTH
  1238. }
  1239. limitRequestPayloadLength = limitRequestPayloadLengthPRNG.Range(
  1240. minLength, maxLength)
  1241. redialTLSProbability = p.Float(parameters.MeekRedialTLSProbability)
  1242. }
  1243. return cookie, limitRequestPayloadLength, redialTLSProbability, nil
  1244. }