meekConn.go 49 KB

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