controller.go 126 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875287628772878287928802881288228832884288528862887288828892890289128922893289428952896289728982899290029012902290329042905290629072908290929102911291229132914291529162917291829192920292129222923292429252926292729282929293029312932293329342935293629372938293929402941294229432944294529462947294829492950295129522953295429552956295729582959296029612962296329642965296629672968296929702971297229732974297529762977297829792980298129822983298429852986298729882989299029912992299329942995299629972998299930003001300230033004300530063007300830093010301130123013301430153016301730183019302030213022302330243025302630273028302930303031303230333034303530363037303830393040304130423043304430453046304730483049305030513052305330543055305630573058305930603061306230633064306530663067306830693070307130723073307430753076307730783079308030813082308330843085308630873088308930903091309230933094309530963097309830993100310131023103310431053106310731083109311031113112311331143115311631173118311931203121312231233124312531263127312831293130313131323133313431353136313731383139314031413142314331443145314631473148314931503151315231533154315531563157315831593160316131623163316431653166316731683169317031713172317331743175317631773178317931803181318231833184318531863187318831893190319131923193319431953196319731983199320032013202320332043205320632073208320932103211321232133214321532163217321832193220322132223223322432253226322732283229323032313232323332343235323632373238323932403241324232433244324532463247324832493250325132523253325432553256325732583259326032613262326332643265326632673268326932703271327232733274327532763277327832793280328132823283328432853286328732883289329032913292329332943295329632973298329933003301330233033304330533063307330833093310331133123313331433153316331733183319332033213322332333243325332633273328332933303331333233333334333533363337333833393340334133423343334433453346334733483349335033513352335333543355335633573358335933603361336233633364336533663367336833693370337133723373337433753376337733783379338033813382338333843385338633873388338933903391339233933394339533963397339833993400340134023403340434053406340734083409341034113412341334143415341634173418341934203421342234233424342534263427342834293430343134323433343434353436343734383439344034413442344334443445344634473448344934503451345234533454345534563457345834593460346134623463346434653466346734683469347034713472347334743475347634773478347934803481348234833484348534863487348834893490349134923493349434953496349734983499350035013502350335043505350635073508350935103511351235133514351535163517351835193520352135223523352435253526352735283529
  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 implements the core tunnel functionality of a Psiphon client.
  20. // The main function is RunForever, which runs a Controller that obtains lists of
  21. // servers, establishes tunnel connections, and runs local proxies through which
  22. // tunneled traffic may be sent.
  23. package psiphon
  24. import (
  25. "context"
  26. "encoding/json"
  27. "fmt"
  28. "math/rand"
  29. "net"
  30. "runtime"
  31. "runtime/pprof"
  32. "sync"
  33. "sync/atomic"
  34. "time"
  35. tls "github.com/Psiphon-Labs/psiphon-tls"
  36. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common"
  37. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/errors"
  38. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/inproxy"
  39. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/parameters"
  40. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/prng"
  41. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/protocol"
  42. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/resolver"
  43. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/tactics"
  44. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/tun"
  45. utls "github.com/Psiphon-Labs/utls"
  46. lrucache "github.com/cognusion/go-cache-lru"
  47. "golang.org/x/time/rate"
  48. )
  49. // Controller is a tunnel lifecycle coordinator. It manages lists of servers to
  50. // connect to; establishes and monitors tunnels; and runs local proxies which
  51. // route traffic through the tunnels.
  52. type Controller struct {
  53. config *Config
  54. runCtx context.Context
  55. stopRunning context.CancelFunc
  56. runWaitGroup *sync.WaitGroup
  57. connectedTunnels chan *Tunnel
  58. failedTunnels chan *Tunnel
  59. tunnelMutex sync.Mutex
  60. establishedOnce bool
  61. tunnelPoolSize int
  62. tunnels []*Tunnel
  63. nextTunnel int
  64. isEstablishing bool
  65. establishStartTime time.Time
  66. protocolSelectionConstraints *protocolSelectionConstraints
  67. concurrentEstablishTunnelsMutex sync.Mutex
  68. establishConnectTunnelCount int
  69. concurrentEstablishTunnels int
  70. concurrentIntensiveEstablishTunnels int
  71. peakConcurrentEstablishTunnels int
  72. peakConcurrentIntensiveEstablishTunnels int
  73. establishInproxyForceSelectionCount int
  74. establishCtx context.Context
  75. stopEstablish context.CancelFunc
  76. establishWaitGroup *sync.WaitGroup
  77. establishedTunnelsCount int32
  78. candidateServerEntries chan *candidateServerEntry
  79. untunneledDialConfig *DialConfig
  80. untunneledSplitTunnelClassifications *lrucache.Cache
  81. signalFetchCommonRemoteServerList chan struct{}
  82. signalFetchObfuscatedServerLists chan struct{}
  83. signalUntunneledDSLFetch chan struct{}
  84. signalTunneledDSLFetch chan struct{}
  85. signalDownloadUpgrade chan string
  86. signalReportServerEntries chan *serverEntriesReportRequest
  87. signalReportConnected chan struct{}
  88. signalRestartEstablishing chan struct{}
  89. serverAffinityDoneBroadcast chan struct{}
  90. packetTunnelClient *tun.Client
  91. packetTunnelTransport *PacketTunnelTransport
  92. staggerMutex sync.Mutex
  93. resolver *resolver.Resolver
  94. steeringIPCache *lrucache.Cache
  95. tlsClientSessionCache utls.ClientSessionCache
  96. quicTLSClientSessionCache tls.ClientSessionCache
  97. inproxyProxyBrokerClientManager *InproxyBrokerClientManager
  98. inproxyClientBrokerClientManager *InproxyBrokerClientManager
  99. inproxyNATStateManager *InproxyNATStateManager
  100. inproxyHandleTacticsMutex sync.Mutex
  101. inproxyLastStoredTactics time.Time
  102. establishSignalForceTacticsFetch chan struct{}
  103. inproxyClientDialRateLimiter *rate.Limiter
  104. currentNetworkMutex sync.Mutex
  105. currentNetworkCtx context.Context
  106. currentNetworkCancelFunc context.CancelFunc
  107. }
  108. // NewController initializes a new controller.
  109. func NewController(config *Config) (controller *Controller, err error) {
  110. if !config.IsCommitted() {
  111. return nil, errors.TraceNew("uncommitted config")
  112. }
  113. // Needed by regen, at least
  114. rand.Seed(int64(time.Now().Nanosecond()))
  115. applyClientAPILevel(config)
  116. // The session ID for the Psiphon server API is used across all
  117. // tunnels established by the controller.
  118. NoticeSessionId(config.SessionID)
  119. if !config.DisableTactics {
  120. // Attempt to apply any valid, local stored tactics. The pre-done context
  121. // ensures no tactics request is attempted now.
  122. doneContext, cancelFunc := context.WithCancel(context.Background())
  123. cancelFunc()
  124. GetTactics(doneContext, config, true)
  125. }
  126. p := config.GetParameters().Get()
  127. splitTunnelClassificationTTL :=
  128. p.Duration(parameters.SplitTunnelClassificationTTL)
  129. splitTunnelClassificationMaxEntries :=
  130. p.Int(parameters.SplitTunnelClassificationMaxEntries)
  131. steeringIPCacheTTL := p.Duration(parameters.SteeringIPCacheTTL)
  132. steeringIPCacheMaxEntries := p.Int(parameters.SteeringIPCacheMaxEntries)
  133. controller = &Controller{
  134. config: config,
  135. runWaitGroup: new(sync.WaitGroup),
  136. // connectedTunnels and failedTunnels buffer sizes are large enough to
  137. // receive full pools of tunnels without blocking. Senders should not block.
  138. connectedTunnels: make(chan *Tunnel, MAX_TUNNEL_POOL_SIZE),
  139. failedTunnels: make(chan *Tunnel, MAX_TUNNEL_POOL_SIZE),
  140. tunnelPoolSize: TUNNEL_POOL_SIZE,
  141. tunnels: make([]*Tunnel, 0),
  142. establishedOnce: false,
  143. isEstablishing: false,
  144. untunneledSplitTunnelClassifications: lrucache.NewWithLRU(
  145. splitTunnelClassificationTTL,
  146. 1*time.Minute,
  147. splitTunnelClassificationMaxEntries),
  148. // TODO: Add a buffer of 1 so we don't miss a signal while receiver is
  149. // starting? Trade-off is potential back-to-back fetch remotes. As-is,
  150. // establish will eventually signal another fetch remote.
  151. signalFetchCommonRemoteServerList: make(chan struct{}),
  152. signalFetchObfuscatedServerLists: make(chan struct{}),
  153. signalUntunneledDSLFetch: make(chan struct{}),
  154. signalTunneledDSLFetch: make(chan struct{}),
  155. signalDownloadUpgrade: make(chan string),
  156. signalReportConnected: make(chan struct{}),
  157. // Using a buffer of 1 to ensure there's no race between the first signal
  158. // sent and a channel receiver initializing; a side effect is that this
  159. // allows 1 additional scan to enqueue while a scan is in progress, possibly
  160. // resulting in one unnecessary scan.
  161. signalReportServerEntries: make(chan *serverEntriesReportRequest, 1),
  162. // signalRestartEstablishing has a buffer of 1 to ensure sending the
  163. // signal doesn't block and receiving won't miss a signal.
  164. signalRestartEstablishing: make(chan struct{}, 1),
  165. steeringIPCache: lrucache.NewWithLRU(
  166. steeringIPCacheTTL,
  167. 1*time.Minute,
  168. steeringIPCacheMaxEntries),
  169. tlsClientSessionCache: utls.NewLRUClientSessionCache(0),
  170. quicTLSClientSessionCache: tls.NewLRUClientSessionCache(0),
  171. }
  172. // Initialize the current network context. This context represents the
  173. // lifetime of the host's current active network interface. When
  174. // Controller.NetworkChanged is called (by the Android and iOS platform
  175. // code), the previous current network interface is considered to be no
  176. // longer active and the corresponding current network context is canceled.
  177. // Components may use currentNetworkCtx to cancel and close old network
  178. // connections and quickly initiate new connections when the active
  179. // interface changes.
  180. controller.currentNetworkCtx, controller.currentNetworkCancelFunc =
  181. context.WithCancel(context.Background())
  182. // Initialize untunneledDialConfig, used by untunneled dials including
  183. // remote server list and upgrade downloads.
  184. controller.untunneledDialConfig = &DialConfig{
  185. UpstreamProxyURL: controller.config.UpstreamProxyURL,
  186. CustomHeaders: controller.config.CustomHeaders,
  187. DeviceBinder: controller.config.deviceBinder,
  188. IPv6Synthesizer: controller.config.IPv6Synthesizer,
  189. ResolveIP: func(ctx context.Context, hostname string) ([]net.IP, error) {
  190. // Note: when domain fronting would be used for untunneled dials a
  191. // copy of untunneledDialConfig should be used instead, which
  192. // redefines ResolveIP such that the corresponding fronting
  193. // provider ID is passed into UntunneledResolveIP to enable the use
  194. // of pre-resolved IPs.
  195. IPs, err := UntunneledResolveIP(
  196. ctx, controller.config, controller.resolver, hostname, "")
  197. if err != nil {
  198. return nil, errors.Trace(err)
  199. }
  200. return IPs, nil
  201. },
  202. TrustedCACertificatesFilename: controller.config.TrustedCACertificatesFilename,
  203. }
  204. if config.PacketTunnelTunFileDescriptor > 0 {
  205. // Run a packet tunnel client. The lifetime of the tun.Client is the
  206. // lifetime of the Controller, so it exists across tunnel establishments
  207. // and reestablishments. The PacketTunnelTransport provides a layer
  208. // that presents a continuosuly existing transport to the tun.Client;
  209. // it's set to use new SSH channels after new SSH tunnel establishes.
  210. packetTunnelTransport := NewPacketTunnelTransport()
  211. packetTunnelClient, err := tun.NewClient(&tun.ClientConfig{
  212. Logger: NoticeCommonLogger(false),
  213. TunFileDescriptor: config.PacketTunnelTunFileDescriptor,
  214. TransparentDNSIPv4Address: config.PacketTunnelTransparentDNSIPv4Address,
  215. TransparentDNSIPv6Address: config.PacketTunnelTransparentDNSIPv6Address,
  216. Transport: packetTunnelTransport,
  217. })
  218. if err != nil {
  219. return nil, errors.Trace(err)
  220. }
  221. controller.packetTunnelClient = packetTunnelClient
  222. controller.packetTunnelTransport = packetTunnelTransport
  223. }
  224. // Initialize shared in-proxy broker clients to be used for all in-proxy
  225. // client dials and in-proxy proxy operations.
  226. //
  227. // Using shared broker connections minimizes the overhead of establishing
  228. // broker connections at the start of an in-proxy dial or operation. By
  229. // design, established broker connections will be retained for up to the
  230. // entire lifetime of the controller run, so past the end of client
  231. // tunnel establishment.
  232. //
  233. // No network operations are performed by NewInproxyBrokerClientManager or
  234. // NewInproxyNATStateManager; each manager operates on demand, when
  235. // in-proxy dials or operations are invoked.
  236. //
  237. // The controller run may include client tunnel establishment, in-proxy
  238. // proxy operations, or both.
  239. //
  240. // Due to the inproxy.InitiatorSessions.NewRoundTrip waitToShareSession
  241. // application-level round trip limitation, there is one broker client
  242. // manager for each of the client and proxy cases, so that neither
  243. // initially blocks while trying to share the others session.
  244. //
  245. // One NAT state manager is shared between both the in-proxy client and
  246. // proxy. While each may have different network discovery policies, any
  247. // discovered network state is valid and useful for both consumers.
  248. // Both broker client and NAT state managers may require resets and update
  249. // when tactics change.
  250. var tacticAppliedReceivers []TacticsAppliedReceiver
  251. isProxy := false
  252. controller.inproxyClientBrokerClientManager = NewInproxyBrokerClientManager(config, isProxy, controller.tlsClientSessionCache)
  253. tacticAppliedReceivers = append(tacticAppliedReceivers, controller.inproxyClientBrokerClientManager)
  254. controller.inproxyNATStateManager = NewInproxyNATStateManager(config)
  255. tacticAppliedReceivers = append(tacticAppliedReceivers, controller.inproxyNATStateManager)
  256. if config.InproxyEnableProxy {
  257. isProxy = true
  258. controller.inproxyProxyBrokerClientManager = NewInproxyBrokerClientManager(config, isProxy, controller.tlsClientSessionCache)
  259. tacticAppliedReceivers = append(tacticAppliedReceivers, controller.inproxyProxyBrokerClientManager)
  260. }
  261. controller.config.SetTacticsAppliedReceivers(tacticAppliedReceivers)
  262. controller.config.SetSignalComponentFailure(controller.SignalComponentFailure)
  263. return controller, nil
  264. }
  265. // Run executes the controller. Run exits if a controller
  266. // component fails or the parent context is canceled.
  267. func (controller *Controller) Run(ctx context.Context) {
  268. if controller.config.LimitCPUThreads {
  269. runtime.GOMAXPROCS(1)
  270. }
  271. pprofRun()
  272. // Ensure fresh repetitive notice state for each run, so the
  273. // client will always get an AvailableEgressRegions notice,
  274. // an initial instance of any repetitive error notice, etc.
  275. ResetRepetitiveNotices()
  276. runCtx, stopRunning := context.WithCancel(ctx)
  277. defer stopRunning()
  278. controller.runCtx = runCtx
  279. controller.stopRunning = stopRunning
  280. // Start components
  281. // Initialize a single resolver to be used by all dials. Sharing a single
  282. // resolver ensures cached results are shared, and that network state
  283. // query overhead is amortized over all dials. Multiple dials can resolve
  284. // domain concurrently.
  285. //
  286. // config.SetResolver makes this resolver available to MakeDialParameters.
  287. controller.resolver = NewResolver(controller.config, true)
  288. defer controller.resolver.Stop()
  289. controller.config.SetResolver(controller.resolver)
  290. // Maintain a cache of steering IPs to be applied to dials. A steering IP
  291. // is an alternate dial IP; for example, steering IPs may be specified by
  292. // a CDN service and used to load balance CDN traffic.
  293. controller.steeringIPCache.Flush()
  294. // TODO: IPv6 support
  295. var listenIP string
  296. if controller.config.ListenInterface == "" {
  297. listenIP = "127.0.0.1"
  298. } else if controller.config.ListenInterface == "any" {
  299. listenIP = "0.0.0.0"
  300. } else {
  301. IPv4Address, _, err := common.GetInterfaceIPAddresses(controller.config.ListenInterface)
  302. if err == nil && IPv4Address == nil {
  303. err = fmt.Errorf("no IPv4 address for interface %s", controller.config.ListenInterface)
  304. }
  305. if err != nil {
  306. NoticeError("error getting listener IP: %v", errors.Trace(err))
  307. return
  308. }
  309. listenIP = IPv4Address.String()
  310. }
  311. // The controller run may include client tunnel establishment, in-proxy
  312. // proxy operations, or both. Local tactics are shared between both modes
  313. // and both modes can fetch tactics.
  314. //
  315. // Limitation: the upgrade downloader is not enabled when client tunnel
  316. // establishment is disabled; upgrade version information is not
  317. // currently distributed to in-proxy proxies
  318. if !controller.config.DisableTunnels {
  319. if !controller.config.DisableLocalSocksProxy {
  320. socksProxy, err := NewSocksProxy(controller.config, controller, listenIP)
  321. if err != nil {
  322. NoticeError("error initializing local SOCKS proxy: %v", errors.Trace(err))
  323. return
  324. }
  325. defer socksProxy.Close()
  326. }
  327. if !controller.config.DisableLocalHTTPProxy {
  328. httpProxy, err := NewHttpProxy(controller.config, controller, listenIP)
  329. if err != nil {
  330. NoticeError("error initializing local HTTP proxy: %v", errors.Trace(err))
  331. return
  332. }
  333. defer httpProxy.Close()
  334. }
  335. if controller.config.EnableUpgradeDownload {
  336. controller.runWaitGroup.Add(1)
  337. go controller.upgradeDownloader()
  338. }
  339. controller.runWaitGroup.Add(1)
  340. go controller.serverEntriesReporter()
  341. controller.runWaitGroup.Add(1)
  342. go controller.connectedReporter()
  343. controller.runWaitGroup.Add(1)
  344. go controller.establishTunnelWatcher()
  345. controller.runWaitGroup.Add(1)
  346. go controller.runTunnels()
  347. if controller.packetTunnelClient != nil {
  348. controller.packetTunnelClient.Start()
  349. }
  350. }
  351. if !controller.config.DisableRemoteServerListFetcher {
  352. if controller.config.RemoteServerListURLs != nil {
  353. controller.runWaitGroup.Add(1)
  354. go controller.remoteServerListFetcher(
  355. "common",
  356. FetchCommonRemoteServerList,
  357. controller.signalFetchCommonRemoteServerList)
  358. }
  359. if controller.config.ObfuscatedServerListRootURLs != nil {
  360. controller.runWaitGroup.Add(1)
  361. go controller.remoteServerListFetcher(
  362. "obfuscated",
  363. FetchObfuscatedServerLists,
  364. controller.signalFetchObfuscatedServerLists)
  365. }
  366. }
  367. if !controller.config.DisableDSLFetcher {
  368. controller.runWaitGroup.Add(1)
  369. go func() {
  370. defer controller.runWaitGroup.Done()
  371. runUntunneledDSLFetcher(
  372. controller.runCtx,
  373. controller.config,
  374. controller.inproxyClientBrokerClientManager,
  375. controller.signalUntunneledDSLFetch)
  376. }()
  377. controller.runWaitGroup.Add(1)
  378. go func() {
  379. defer controller.runWaitGroup.Done()
  380. runTunneledDSLFetcher(
  381. controller.runCtx,
  382. controller.config,
  383. controller.getNextActiveTunnel,
  384. controller.signalTunneledDSLFetch)
  385. }()
  386. }
  387. if controller.config.InproxyEnableProxy {
  388. controller.runWaitGroup.Add(1)
  389. go controller.runInproxyProxy()
  390. }
  391. // Wait while running
  392. <-controller.runCtx.Done()
  393. NoticeInfo("controller stopped")
  394. // To assist with diagnosing unexpected shutdown hangs, log a goroutine
  395. // profile if the wait operation runs over a deadline. This diagnostic
  396. // goroutine is intentially not awaited on.
  397. signalDoneShutdown := make(chan struct{})
  398. go func() {
  399. deadlineSeconds := 60
  400. if controller.config.ShutdownGoroutineProfileDeadlineSeconds != nil {
  401. deadlineSeconds = *controller.config.ShutdownGoroutineProfileDeadlineSeconds
  402. }
  403. if deadlineSeconds == 0 {
  404. return
  405. }
  406. timer := time.NewTimer(time.Duration(deadlineSeconds) * time.Second)
  407. defer timer.Stop()
  408. select {
  409. case <-signalDoneShutdown:
  410. return
  411. case <-timer.C:
  412. }
  413. pprof.Lookup("goroutine").WriteTo(
  414. NewNoticeLineWriter("Goroutine"), 1)
  415. }()
  416. // Shutdown
  417. if controller.packetTunnelClient != nil {
  418. controller.packetTunnelClient.Stop()
  419. }
  420. // Cleanup current network context
  421. controller.currentNetworkCancelFunc()
  422. // All workers -- runTunnels, establishment workers, and auxilliary
  423. // workers such as fetch remote server list and untunneled uprade
  424. // download -- operate with the controller run context and will all
  425. // be interrupted when the run context is done.
  426. controller.runWaitGroup.Wait()
  427. close(signalDoneShutdown)
  428. NoticeInfo("exiting controller")
  429. NoticeExiting()
  430. }
  431. // SignalComponentFailure notifies the controller that an associated component has failed.
  432. // This will terminate the controller.
  433. func (controller *Controller) SignalComponentFailure() {
  434. NoticeWarning("controller shutdown due to component failure")
  435. controller.stopRunning()
  436. }
  437. // SetDynamicConfig overrides the sponsor ID and authorizations fields of the
  438. // Controller config with the input values. The new values will be used in the
  439. // next tunnel connection.
  440. func (controller *Controller) SetDynamicConfig(sponsorID string, authorizations []string) {
  441. controller.config.SetDynamicConfig(sponsorID, authorizations)
  442. }
  443. // NetworkChanged initiates a reset of all open network connections, including
  444. // a tunnel reconnect, if one is running, as well as terminating any in-proxy
  445. // proxy connections.
  446. func (controller *Controller) NetworkChanged() {
  447. // Explicitly reset components that don't use the current network context.
  448. controller.TerminateNextActiveTunnel()
  449. if controller.inproxyProxyBrokerClientManager != nil {
  450. err := controller.inproxyProxyBrokerClientManager.NetworkChanged()
  451. if err != nil {
  452. NoticeError("NetworkChanged failed: %v", errors.Trace(err))
  453. // Log and continue running.
  454. }
  455. }
  456. err := controller.inproxyClientBrokerClientManager.NetworkChanged()
  457. if err != nil {
  458. NoticeError("NetworkChanged failed: %v", errors.Trace(err))
  459. // Log and continue running.
  460. }
  461. controller.config.networkIDGetter.FlushCache()
  462. // Cancel the previous current network context, which will interrupt any
  463. // operations using this context. Then create a new context for the new
  464. // current network.
  465. controller.currentNetworkMutex.Lock()
  466. defer controller.currentNetworkMutex.Unlock()
  467. controller.currentNetworkCancelFunc()
  468. controller.currentNetworkCtx, controller.currentNetworkCancelFunc =
  469. context.WithCancel(context.Background())
  470. }
  471. func (controller *Controller) getCurrentNetworkContext() context.Context {
  472. controller.currentNetworkMutex.Lock()
  473. defer controller.currentNetworkMutex.Unlock()
  474. return controller.currentNetworkCtx
  475. }
  476. // TerminateNextActiveTunnel terminates the active tunnel, which will initiate
  477. // establishment of a new tunnel.
  478. func (controller *Controller) TerminateNextActiveTunnel() {
  479. tunnel := controller.getNextActiveTunnel()
  480. if tunnel != nil {
  481. controller.SignalTunnelFailure(tunnel)
  482. NoticeInfo("terminated tunnel: %s", tunnel.dialParams.ServerEntry.GetDiagnosticID())
  483. }
  484. }
  485. // ExportExchangePayload creates a payload for client-to-client server
  486. // connection info exchange. See the comment for psiphon.ExportExchangePayload
  487. // for more details.
  488. func (controller *Controller) ExportExchangePayload() string {
  489. return ExportExchangePayload(controller.config)
  490. }
  491. // ImportExchangePayload imports a payload generated by ExportExchangePayload.
  492. // See the comment for psiphon.ImportExchangePayload for more details about
  493. // the import.
  494. //
  495. // When the import is successful, a signal is set to trigger a restart any
  496. // establishment in progress. This will cause the newly imported server entry
  497. // to be prioritized, which it otherwise would not be in later establishment
  498. // rounds. The establishment process continues after ImportExchangePayload
  499. // returns.
  500. //
  501. // If the client already has a connected tunnel, or a tunnel connection is
  502. // established concurrently with the import, the signal has no effect as the
  503. // overall goal is establish _any_ connection.
  504. func (controller *Controller) ImportExchangePayload(payload string) bool {
  505. // Race condition: if a new tunnel connection is established concurrently
  506. // with the import, either that tunnel's server entry of the imported server
  507. // entry may end up as the affinity server.
  508. ok := ImportExchangePayload(controller.config, payload)
  509. if !ok {
  510. return false
  511. }
  512. select {
  513. case controller.signalRestartEstablishing <- struct{}{}:
  514. default:
  515. }
  516. return true
  517. }
  518. // remoteServerListFetcher fetches an out-of-band list of server entries
  519. // for more tunnel candidates. It fetches when signalled, with retries
  520. // on failure.
  521. func (controller *Controller) remoteServerListFetcher(
  522. name string,
  523. fetcher RemoteServerListFetcher,
  524. signal <-chan struct{}) {
  525. defer controller.runWaitGroup.Done()
  526. var lastFetchTime time.Time
  527. fetcherLoop:
  528. for {
  529. // Wait for a signal before fetching
  530. select {
  531. case <-signal:
  532. case <-controller.runCtx.Done():
  533. break fetcherLoop
  534. }
  535. // Skip fetch entirely (i.e., send no request at all, even when ETag would save
  536. // on response size) when a recent fetch was successful
  537. stalePeriod := controller.config.GetParameters().Get().Duration(
  538. parameters.FetchRemoteServerListStalePeriod)
  539. if !lastFetchTime.IsZero() &&
  540. lastFetchTime.Add(stalePeriod).After(time.Now()) {
  541. continue
  542. }
  543. retryLoop:
  544. for attempt := 0; ; attempt++ {
  545. // Don't attempt to fetch while there is no network connectivity,
  546. // to avoid alert notice noise.
  547. if !WaitForNetworkConnectivity(
  548. controller.runCtx,
  549. controller.config.NetworkConnectivityChecker,
  550. nil) {
  551. break fetcherLoop
  552. }
  553. // Pick any active tunnel and make the next fetch attempt. If there's
  554. // no active tunnel, the untunneledDialConfig will be used.
  555. tunnel := controller.getNextActiveTunnel()
  556. err := fetcher(
  557. controller.runCtx,
  558. controller.config,
  559. attempt,
  560. tunnel,
  561. controller.untunneledDialConfig,
  562. controller.tlsClientSessionCache)
  563. if err == nil {
  564. lastFetchTime = time.Now()
  565. break retryLoop
  566. }
  567. NoticeWarning("failed to fetch %s remote server list: %v",
  568. name, errors.Trace(err))
  569. retryPeriod := controller.config.GetParameters().Get().Duration(
  570. parameters.FetchRemoteServerListRetryPeriod)
  571. timer := time.NewTimer(retryPeriod)
  572. select {
  573. case <-timer.C:
  574. case <-controller.runCtx.Done():
  575. timer.Stop()
  576. break fetcherLoop
  577. }
  578. }
  579. }
  580. NoticeInfo("exiting %s remote server list fetcher", name)
  581. }
  582. // upgradeDownloader makes periodic attempts to complete a client upgrade
  583. // download. DownloadUpgrade() is resumable, so each attempt has potential for
  584. // getting closer to completion, even in conditions where the download or
  585. // tunnel is repeatedly interrupted.
  586. // An upgrade download is triggered by either a handshake response indicating
  587. // that a new version is available; or after failing to connect, in which case
  588. // it's useful to check, out-of-band, for an upgrade with new circumvention
  589. // capabilities.
  590. // Once the download operation completes successfully, the downloader exits
  591. // and is not run again: either there is not a newer version, or the upgrade
  592. // has been downloaded and is ready to be applied.
  593. // We're assuming that the upgrade will be applied and the entire system
  594. // restarted before another upgrade is to be downloaded.
  595. //
  596. // TODO: refactor upgrade downloader and remote server list fetcher to use
  597. // common code (including the resumable download routines).
  598. func (controller *Controller) upgradeDownloader() {
  599. defer controller.runWaitGroup.Done()
  600. var lastDownloadTime time.Time
  601. downloadLoop:
  602. for {
  603. // Wait for a signal before downloading
  604. var handshakeVersion string
  605. select {
  606. case handshakeVersion = <-controller.signalDownloadUpgrade:
  607. case <-controller.runCtx.Done():
  608. break downloadLoop
  609. }
  610. stalePeriod := controller.config.GetParameters().Get().Duration(
  611. parameters.FetchUpgradeStalePeriod)
  612. // Unless handshake is explicitly advertizing a new version, skip
  613. // checking entirely when a recent download was successful.
  614. if handshakeVersion == "" &&
  615. !lastDownloadTime.IsZero() &&
  616. lastDownloadTime.Add(stalePeriod).After(time.Now()) {
  617. continue
  618. }
  619. retryLoop:
  620. for attempt := 0; ; attempt++ {
  621. // Don't attempt to download while there is no network connectivity,
  622. // to avoid alert notice noise.
  623. if !WaitForNetworkConnectivity(
  624. controller.runCtx,
  625. controller.config.NetworkConnectivityChecker,
  626. nil) {
  627. break downloadLoop
  628. }
  629. // Pick any active tunnel and make the next download attempt. If there's
  630. // no active tunnel, the untunneledDialConfig will be used.
  631. tunnel := controller.getNextActiveTunnel()
  632. err := DownloadUpgrade(
  633. controller.runCtx,
  634. controller.config,
  635. attempt,
  636. handshakeVersion,
  637. tunnel,
  638. controller.untunneledDialConfig,
  639. controller.tlsClientSessionCache)
  640. if err == nil {
  641. lastDownloadTime = time.Now()
  642. break retryLoop
  643. }
  644. NoticeWarning("failed to download upgrade: %v", errors.Trace(err))
  645. timeout := controller.config.GetParameters().Get().Duration(
  646. parameters.FetchUpgradeRetryPeriod)
  647. timer := time.NewTimer(timeout)
  648. select {
  649. case <-timer.C:
  650. case <-controller.runCtx.Done():
  651. timer.Stop()
  652. break downloadLoop
  653. }
  654. }
  655. }
  656. NoticeInfo("exiting upgrade downloader")
  657. }
  658. type serverEntriesReportRequest struct {
  659. constraints *protocolSelectionConstraints
  660. }
  661. // serverEntriesReporter performs scans over all server entries to report on
  662. // available tunnel candidates, subject to protocol selection constraints, and
  663. // available egress regions.
  664. //
  665. // Because scans may be slow, depending on the client device and server entry
  666. // list size, serverEntriesReporter is used to perform asychronous, background
  667. // operations that would otherwise block establishment. This includes emitting
  668. // diagnotic notices that are informational (CandidateServers) or which do not
  669. // need to emit before establishment starts (AvailableEgressRegions).
  670. //
  671. // serverEntriesReporter also serves to combine these scans, which would
  672. // otherwise be logically independent, due to the performance impact of scans.
  673. //
  674. // Limitation: The underlying datastore implementation _may_ block write
  675. // transactions while there are open read transactions. For example, bolt
  676. // write transactions which need to re-map the data file (when the datastore
  677. // grows) will block on open read transactions. In these scenarios, a slow
  678. // scan will still block other operations.
  679. //
  680. // serverEntriesReporter runs beyond the establishment phase, since it's
  681. // important for notices such as AvailableEgressRegions to eventually emit
  682. // even if already established. serverEntriesReporter scans are cancellable,
  683. // so controller shutdown is not blocked by slow scans.
  684. func (controller *Controller) serverEntriesReporter() {
  685. defer controller.runWaitGroup.Done()
  686. loop:
  687. for {
  688. var request *serverEntriesReportRequest
  689. select {
  690. case request = <-controller.signalReportServerEntries:
  691. case <-controller.runCtx.Done():
  692. break loop
  693. }
  694. egressRegion := controller.config.EgressRegion
  695. constraints := request.constraints
  696. regions := make(map[string]bool)
  697. initialCandidates := 0
  698. candidates := 0
  699. callback := func(serverEntry *protocol.ServerEntry) bool {
  700. // In establishment, excludeIntensive depends on what set of protocols are
  701. // already being dialed. For these reports, don't exclude intensive
  702. // protocols as any intensive candidate can always be an available
  703. // candidate at some point.
  704. excludeIntensive := false
  705. isInitialCandidate := constraints.isInitialCandidate(excludeIntensive, serverEntry)
  706. isCandidate := constraints.isCandidate(excludeIntensive, serverEntry)
  707. if egressRegion == "" || serverEntry.Region == egressRegion {
  708. if isInitialCandidate {
  709. initialCandidates += 1
  710. }
  711. if isCandidate {
  712. candidates += 1
  713. }
  714. }
  715. isAvailable := isCandidate
  716. if constraints.hasInitialProtocols() {
  717. // Available egress regions is subject to an initial limit constraint, if
  718. // present: see AvailableEgressRegions comment in launchEstablishing.
  719. isAvailable = isInitialCandidate
  720. }
  721. if isAvailable {
  722. // Ignore server entries with no region field.
  723. if serverEntry.Region != "" {
  724. regions[serverEntry.Region] = true
  725. }
  726. }
  727. select {
  728. case <-controller.runCtx.Done():
  729. // Don't block controller shutdown: cancel the scan.
  730. return false
  731. default:
  732. return true
  733. }
  734. }
  735. startTime := time.Now()
  736. err := ScanServerEntries(callback)
  737. if err != nil {
  738. NoticeWarning("ScanServerEntries failed: %v", errors.Trace(err))
  739. continue
  740. }
  741. // Report this duration in CandidateServers as an indication of datastore
  742. // performance.
  743. duration := time.Since(startTime)
  744. NoticeCandidateServers(
  745. controller.config.EgressRegion,
  746. constraints,
  747. initialCandidates,
  748. candidates,
  749. duration)
  750. availableEgressRegions := make([]string, 0, len(regions))
  751. for region := range regions {
  752. availableEgressRegions = append(availableEgressRegions, region)
  753. }
  754. NoticeAvailableEgressRegions(
  755. availableEgressRegions)
  756. }
  757. NoticeInfo("exiting server entries reporter")
  758. }
  759. // signalServerEntriesReporter triggers a new server entry report.The report
  760. // is considered to be informational and may or may not run, depending on
  761. // whether another run is already in progress.
  762. func (controller *Controller) signalServerEntriesReporter(
  763. request *serverEntriesReportRequest) {
  764. select {
  765. case controller.signalReportServerEntries <- request:
  766. default:
  767. }
  768. }
  769. // connectedReporter sends periodic "connected" requests to the Psiphon API.
  770. // These requests are for server-side unique user stats calculation. See the
  771. // comment in DoConnectedRequest for a description of the request mechanism.
  772. //
  773. // To correctly count daily unique users, only one connected request is made
  774. // across all simultaneous multi-tunnels; and the connected request is
  775. // repeated every 24h.
  776. //
  777. // The signalReportConnected mechanism is used to trigger a connected request
  778. // immediately after a reconnect. While strictly only one connected request
  779. // per 24h is required in order to count daily unique users, the connected
  780. // request also delivers the establishment duration metric (which includes
  781. // time elapsed performing the handshake request) and additional fragmentation
  782. // metrics; these metrics are measured for each tunnel.
  783. func (controller *Controller) connectedReporter() {
  784. defer controller.runWaitGroup.Done()
  785. // session is nil when DisableApi is set
  786. if controller.config.DisableApi {
  787. return
  788. }
  789. select {
  790. case <-controller.signalReportConnected:
  791. // Make the initial connected request
  792. case <-controller.runCtx.Done():
  793. return
  794. }
  795. loop:
  796. for {
  797. // Pick any active tunnel and make the next connected request. No error is
  798. // logged if there's no active tunnel, as that's not an unexpected
  799. // condition.
  800. reported := false
  801. tunnel := controller.getNextActiveTunnel()
  802. if tunnel != nil {
  803. err := tunnel.serverContext.DoConnectedRequest()
  804. if err == nil {
  805. reported = true
  806. } else {
  807. NoticeWarning("failed to make connected request: %v",
  808. errors.Trace(err))
  809. }
  810. }
  811. // Schedule the next connected request and wait. This duration is not a
  812. // dynamic ClientParameter as the daily unique user stats logic specifically
  813. // requires a "connected" request no more or less often than every 24h.
  814. var duration time.Duration
  815. if reported {
  816. duration = 24 * time.Hour
  817. } else {
  818. duration = controller.config.GetParameters().Get().Duration(
  819. parameters.PsiphonAPIConnectedRequestRetryPeriod)
  820. }
  821. timer := time.NewTimer(duration)
  822. doBreak := false
  823. select {
  824. case <-controller.signalReportConnected:
  825. case <-timer.C:
  826. // Make another connected request
  827. case <-controller.runCtx.Done():
  828. doBreak = true
  829. }
  830. timer.Stop()
  831. if doBreak {
  832. break loop
  833. }
  834. }
  835. NoticeInfo("exiting connected reporter")
  836. }
  837. func (controller *Controller) signalConnectedReporter() {
  838. // session is nil when DisableApi is set
  839. if controller.config.DisableApi {
  840. return
  841. }
  842. select {
  843. case controller.signalReportConnected <- struct{}{}:
  844. default:
  845. }
  846. }
  847. // establishTunnelWatcher terminates the controller if a tunnel
  848. // has not been established in the configured time period. This
  849. // is regardless of how many tunnels are presently active -- meaning
  850. // that if an active tunnel was established and lost the controller
  851. // is left running (to re-establish).
  852. func (controller *Controller) establishTunnelWatcher() {
  853. defer controller.runWaitGroup.Done()
  854. timeout := controller.config.GetParameters().Get().Duration(
  855. parameters.EstablishTunnelTimeout)
  856. if timeout > 0 {
  857. timer := time.NewTimer(timeout)
  858. defer timer.Stop()
  859. select {
  860. case <-timer.C:
  861. if !controller.hasEstablishedOnce() {
  862. NoticeEstablishTunnelTimeout(timeout)
  863. controller.SignalComponentFailure()
  864. }
  865. case <-controller.runCtx.Done():
  866. }
  867. }
  868. NoticeInfo("exiting establish tunnel watcher")
  869. }
  870. // runTunnels is the controller tunnel management main loop. It starts and stops
  871. // establishing tunnels based on the target tunnel pool size and the current size
  872. // of the pool. Tunnels are established asynchronously using worker goroutines.
  873. //
  874. // When there are no server entries for the target region/protocol, the
  875. // establishCandidateGenerator will yield no candidates and wait before
  876. // trying again. In the meantime, a remote server entry fetch may supply
  877. // valid candidates.
  878. //
  879. // When a tunnel is established, it's added to the active pool. The tunnel's
  880. // operateTunnel goroutine monitors the tunnel.
  881. //
  882. // When a tunnel fails, it's removed from the pool and the establish process is
  883. // restarted to fill the pool.
  884. func (controller *Controller) runTunnels() {
  885. defer controller.runWaitGroup.Done()
  886. // Start running
  887. controller.startEstablishing()
  888. loop:
  889. for {
  890. select {
  891. case <-controller.signalRestartEstablishing:
  892. // signalRestartEstablishing restarts any establishment in progress. One
  893. // use case for this is to prioritize a newly imported, exchanged server
  894. // entry, which will be in the affinity position.
  895. //
  896. // It's possible for another connection to establish concurrent to signalling;
  897. // since the overall goal remains to establish _any_ connection, we accept that
  898. // in some cases the exchanged server entry may not get used.
  899. if controller.isEstablishing {
  900. controller.stopEstablishing()
  901. controller.startEstablishing()
  902. }
  903. case failedTunnel := <-controller.failedTunnels:
  904. NoticeWarning("tunnel failed: %s", failedTunnel.dialParams.ServerEntry.GetDiagnosticID())
  905. controller.terminateTunnel(failedTunnel)
  906. // Clear the reference to this tunnel before calling startEstablishing,
  907. // which will invoke a garbage collection.
  908. failedTunnel = nil
  909. // Concurrency note: only this goroutine may call startEstablishing/stopEstablishing,
  910. // which reference controller.isEstablishing.
  911. controller.startEstablishing()
  912. case connectedTunnel := <-controller.connectedTunnels:
  913. // Tunnel establishment has two phases: connection and activation.
  914. //
  915. // Connection is run concurrently by the establishTunnelWorkers, to minimize
  916. // delay when it's not yet known which server and protocol will be available
  917. // and unblocked.
  918. //
  919. // Activation is run serially, here, to minimize the overhead of making a
  920. // handshake request and starting the operateTunnel management worker for a
  921. // tunnel which may be discarded.
  922. //
  923. // When the active tunnel will complete establishment, establishment is
  924. // stopped before activation. This interrupts all connecting tunnels and
  925. // garbage collects their memory. The purpose is to minimize memory
  926. // pressure when the handshake request is made. In the unlikely case that the
  927. // handshake fails, establishment is restarted.
  928. //
  929. // Any delays in stopEstablishing will delay the handshake for the last
  930. // active tunnel.
  931. //
  932. // In the typical case of tunnelPoolSize of 1, only a single handshake is
  933. // performed and the homepages notices file, when used, will not be modifed
  934. // after the NoticeTunnels(1) [i.e., connected] until NoticeTunnels(0) [i.e.,
  935. // disconnected]. For tunnelPoolSize > 1, serial handshakes only ensures that
  936. // each set of emitted NoticeHomepages is contiguous.
  937. active, outstanding := controller.numTunnels()
  938. // discardTunnel will be true here when already fully established.
  939. discardTunnel := (outstanding <= 0)
  940. isFirstTunnel := (active == 0)
  941. isLastTunnel := (outstanding == 1)
  942. if !discardTunnel {
  943. if isLastTunnel {
  944. controller.stopEstablishing()
  945. }
  946. // In the case of multi-tunnels, only the first tunnel will send status requests,
  947. // including transfer stats (domain bytes), persistent stats, and prune checks.
  948. // While transfer stats and persistent stats use a "take out" scheme that would
  949. // allow for multiple, concurrent requesters, the prune check does not.
  950. isStatusReporter := isFirstTunnel
  951. err := connectedTunnel.Activate(
  952. controller.runCtx, controller, isStatusReporter)
  953. if err != nil {
  954. NoticeWarning("failed to activate %s: %v",
  955. connectedTunnel.dialParams.ServerEntry.GetDiagnosticID(),
  956. errors.Trace(err))
  957. discardTunnel = true
  958. } else {
  959. // It's unlikely that registerTunnel will fail, since only this goroutine
  960. // calls registerTunnel -- and after checking numTunnels; so failure is not
  961. // expected.
  962. if !controller.registerTunnel(connectedTunnel) {
  963. NoticeWarning("failed to register %s: %v",
  964. connectedTunnel.dialParams.ServerEntry.GetDiagnosticID(),
  965. errors.Trace(err))
  966. discardTunnel = true
  967. }
  968. }
  969. // May need to replace this tunnel
  970. if isLastTunnel && discardTunnel {
  971. controller.startEstablishing()
  972. }
  973. }
  974. if discardTunnel {
  975. controller.discardTunnel(connectedTunnel)
  976. // Clear the reference to this discarded tunnel and immediately run
  977. // a garbage collection to reclaim its memory.
  978. connectedTunnel = nil
  979. DoGarbageCollection()
  980. // Skip the rest of this case
  981. break
  982. }
  983. atomic.AddInt32(&controller.establishedTunnelsCount, 1)
  984. NoticeActiveTunnel(
  985. connectedTunnel.dialParams.ServerEntry.GetDiagnosticID(),
  986. connectedTunnel.dialParams.TunnelProtocol)
  987. NoticeConnectedServerRegion(connectedTunnel.dialParams.ServerEntry.Region)
  988. if isFirstTunnel {
  989. // Signal a connected request on each 1st tunnel establishment. For
  990. // multi-tunnels, the session is connected as long as at least one
  991. // tunnel is established.
  992. controller.signalConnectedReporter()
  993. // Signal a tunneled DSL fetch. The tunneled fetch is similar
  994. // to the handshake API discovery mechanism:
  995. // opportunistically distribute a small number of new server
  996. // entries to clients that are already able to connect.
  997. select {
  998. case controller.signalTunneledDSLFetch <- struct{}{}:
  999. default:
  1000. }
  1001. // If the handshake indicated that a new client version is available,
  1002. // trigger an upgrade download.
  1003. // Note: serverContext is nil when DisableApi is set
  1004. if connectedTunnel.serverContext != nil &&
  1005. connectedTunnel.serverContext.clientUpgradeVersion != "" {
  1006. handshakeVersion := connectedTunnel.serverContext.clientUpgradeVersion
  1007. select {
  1008. case controller.signalDownloadUpgrade <- handshakeVersion:
  1009. default:
  1010. }
  1011. }
  1012. }
  1013. // Set the new tunnel as the transport for the packet tunnel. The packet tunnel
  1014. // client remains up when reestablishing, but no packets are relayed while there
  1015. // is no connected tunnel. UseTunnel will establish a new packet tunnel SSH
  1016. // channel over the new SSH tunnel and configure the packet tunnel client to use
  1017. // the new SSH channel as its transport.
  1018. //
  1019. // Note: as is, this logic is suboptimal for tunnelPoolSize > 1, as this would
  1020. // continuously initialize new packet tunnel sessions for each established
  1021. // server. For now, config validation requires tunnelPoolSize == 1 when
  1022. // the packet tunnel is used.
  1023. if controller.packetTunnelTransport != nil {
  1024. controller.packetTunnelTransport.UseTunnel(connectedTunnel)
  1025. }
  1026. if controller.isFullyEstablished() {
  1027. controller.stopEstablishing()
  1028. }
  1029. case <-controller.runCtx.Done():
  1030. break loop
  1031. }
  1032. }
  1033. // Stop running
  1034. controller.stopEstablishing()
  1035. controller.terminateAllTunnels()
  1036. // Drain tunnel channels
  1037. close(controller.connectedTunnels)
  1038. for tunnel := range controller.connectedTunnels {
  1039. controller.discardTunnel(tunnel)
  1040. }
  1041. close(controller.failedTunnels)
  1042. for tunnel := range controller.failedTunnels {
  1043. controller.discardTunnel(tunnel)
  1044. }
  1045. NoticeInfo("exiting run tunnels")
  1046. }
  1047. // SignalSeededNewSLOK implements the TunnelOwner interface. This function
  1048. // is called by Tunnel.operateTunnel when the tunnel has received a new,
  1049. // previously unknown SLOK from the server. The Controller triggers an OSL
  1050. // fetch, as the new SLOK may be sufficient to access new OSLs.
  1051. func (controller *Controller) SignalSeededNewSLOK() {
  1052. select {
  1053. case controller.signalFetchObfuscatedServerLists <- struct{}{}:
  1054. default:
  1055. }
  1056. // Reset any delay for the next tunneled DSL request. The next time a
  1057. // tunnel connects, the DSL request will launch, and the fetcher will
  1058. // attempt to reassemble OSLs, now with this new SLOK.
  1059. //
  1060. // The delay for the next untunneled DSL request is not reset since that
  1061. // request typically fetches many more server entries, which is more
  1062. // appropriate for when a client is unable to connect. Receiving a new
  1063. // SLOK implies the client is currently connected and is likely to
  1064. // reconnect again and arrive at the tunneled DSL request.
  1065. //
  1066. // TODO: launch an immediate tunneled DSL request?
  1067. _ = DSLSetLastTunneledFetchTime(time.Time{})
  1068. }
  1069. // SignalTunnelFailure implements the TunnelOwner interface. This function
  1070. // is called by Tunnel.operateTunnel when the tunnel has detected that it
  1071. // has failed. The Controller will signal runTunnels to create a new
  1072. // tunnel and/or remove the tunnel from the list of active tunnels.
  1073. func (controller *Controller) SignalTunnelFailure(tunnel *Tunnel) {
  1074. // Don't block. Assumes the receiver has a buffer large enough for
  1075. // the typical number of operated tunnels. In case there's no room,
  1076. // terminate the tunnel (runTunnels won't get a signal in this case,
  1077. // but the tunnel will be removed from the list of active tunnels).
  1078. select {
  1079. case controller.failedTunnels <- tunnel:
  1080. default:
  1081. controller.terminateTunnel(tunnel)
  1082. }
  1083. }
  1084. // discardTunnel disposes of a successful connection that is no longer required.
  1085. func (controller *Controller) discardTunnel(tunnel *Tunnel) {
  1086. NoticeInfo("discard tunnel: %s", tunnel.dialParams.ServerEntry.GetDiagnosticID())
  1087. // TODO: not calling PromoteServerEntry, since that would rank the
  1088. // discarded tunnel before fully active tunnels. Can a discarded tunnel
  1089. // be promoted (since it connects), but with lower rank than all active
  1090. // tunnels?
  1091. tunnel.Close(true)
  1092. }
  1093. // registerTunnel adds the connected tunnel to the pool of active tunnels
  1094. // which are candidates for port forwarding. Returns true if the pool has an
  1095. // empty slot and false if the pool is full (caller should discard the tunnel).
  1096. func (controller *Controller) registerTunnel(tunnel *Tunnel) bool {
  1097. controller.tunnelMutex.Lock()
  1098. defer controller.tunnelMutex.Unlock()
  1099. if len(controller.tunnels) >= controller.tunnelPoolSize {
  1100. return false
  1101. }
  1102. // Perform a final check just in case we've established
  1103. // a duplicate connection.
  1104. for _, activeTunnel := range controller.tunnels {
  1105. if activeTunnel.dialParams.ServerEntry.IpAddress ==
  1106. tunnel.dialParams.ServerEntry.IpAddress {
  1107. NoticeWarning("duplicate tunnel: %s", tunnel.dialParams.ServerEntry.GetDiagnosticID())
  1108. return false
  1109. }
  1110. }
  1111. controller.establishedOnce = true
  1112. controller.tunnels = append(controller.tunnels, tunnel)
  1113. NoticeTunnels(len(controller.tunnels))
  1114. // Promote this successful tunnel to first rank so it's one
  1115. // of the first candidates next time establish runs.
  1116. // Connecting to a TargetServerEntry does not change the
  1117. // ranking.
  1118. if controller.config.TargetServerEntry == "" {
  1119. err := PromoteServerEntry(controller.config, tunnel.dialParams.ServerEntry.IpAddress)
  1120. if err != nil {
  1121. NoticeWarning("PromoteServerEntry failed: %v", errors.Trace(err))
  1122. // Proceed with using tunnel
  1123. }
  1124. }
  1125. return true
  1126. }
  1127. // hasEstablishedOnce indicates if at least one active tunnel has
  1128. // been established up to this point. This is regardeless of how many
  1129. // tunnels are presently active.
  1130. func (controller *Controller) hasEstablishedOnce() bool {
  1131. controller.tunnelMutex.Lock()
  1132. defer controller.tunnelMutex.Unlock()
  1133. return controller.establishedOnce
  1134. }
  1135. // isFullyEstablished indicates if the pool of active tunnels is full.
  1136. func (controller *Controller) isFullyEstablished() bool {
  1137. controller.tunnelMutex.Lock()
  1138. defer controller.tunnelMutex.Unlock()
  1139. return len(controller.tunnels) >= controller.tunnelPoolSize
  1140. }
  1141. // awaitFullyEstablished blocks until isFullyEstablished is true or the
  1142. // controller run ends.
  1143. func (controller *Controller) awaitFullyEstablished() bool {
  1144. // TODO: don't poll, add a signal
  1145. ticker := time.NewTicker(1 * time.Second)
  1146. defer ticker.Stop()
  1147. for {
  1148. if controller.isFullyEstablished() {
  1149. return true
  1150. }
  1151. select {
  1152. case <-ticker.C:
  1153. // Check isFullyEstablished again
  1154. case <-controller.runCtx.Done():
  1155. return false
  1156. }
  1157. }
  1158. }
  1159. // numTunnels returns the number of active and outstanding tunnels.
  1160. // Oustanding is the number of tunnels required to fill the pool of
  1161. // active tunnels.
  1162. func (controller *Controller) numTunnels() (int, int) {
  1163. controller.tunnelMutex.Lock()
  1164. defer controller.tunnelMutex.Unlock()
  1165. active := len(controller.tunnels)
  1166. outstanding := controller.tunnelPoolSize - len(controller.tunnels)
  1167. return active, outstanding
  1168. }
  1169. // terminateTunnel removes a tunnel from the pool of active tunnels
  1170. // and closes the tunnel. The next-tunnel state used by getNextActiveTunnel
  1171. // is adjusted as required.
  1172. func (controller *Controller) terminateTunnel(tunnel *Tunnel) {
  1173. controller.tunnelMutex.Lock()
  1174. defer controller.tunnelMutex.Unlock()
  1175. for index, activeTunnel := range controller.tunnels {
  1176. if tunnel == activeTunnel {
  1177. controller.tunnels = append(
  1178. controller.tunnels[:index], controller.tunnels[index+1:]...)
  1179. if controller.nextTunnel > index {
  1180. controller.nextTunnel--
  1181. }
  1182. if controller.nextTunnel >= len(controller.tunnels) {
  1183. controller.nextTunnel = 0
  1184. }
  1185. activeTunnel.Close(false)
  1186. NoticeTunnels(len(controller.tunnels))
  1187. break
  1188. }
  1189. }
  1190. }
  1191. // terminateAllTunnels empties the tunnel pool, closing all active tunnels.
  1192. // This is used when shutting down the controller.
  1193. func (controller *Controller) terminateAllTunnels() {
  1194. controller.tunnelMutex.Lock()
  1195. defer controller.tunnelMutex.Unlock()
  1196. // Closing all tunnels in parallel. In an orderly shutdown, each tunnel
  1197. // may take a few seconds to send a final status request. We only want
  1198. // to wait as long as the single slowest tunnel.
  1199. closeWaitGroup := new(sync.WaitGroup)
  1200. closeWaitGroup.Add(len(controller.tunnels))
  1201. for _, activeTunnel := range controller.tunnels {
  1202. tunnel := activeTunnel
  1203. go func() {
  1204. defer closeWaitGroup.Done()
  1205. tunnel.Close(false)
  1206. }()
  1207. }
  1208. closeWaitGroup.Wait()
  1209. controller.tunnels = make([]*Tunnel, 0)
  1210. controller.nextTunnel = 0
  1211. NoticeTunnels(len(controller.tunnels))
  1212. }
  1213. // getNextActiveTunnel returns the next tunnel from the pool of active
  1214. // tunnels. Currently, tunnel selection order is simple round-robin.
  1215. func (controller *Controller) getNextActiveTunnel() (tunnel *Tunnel) {
  1216. controller.tunnelMutex.Lock()
  1217. defer controller.tunnelMutex.Unlock()
  1218. if len(controller.tunnels) == 0 {
  1219. return nil
  1220. }
  1221. tunnel = controller.tunnels[controller.nextTunnel]
  1222. controller.nextTunnel =
  1223. (controller.nextTunnel + 1) % len(controller.tunnels)
  1224. return tunnel
  1225. }
  1226. // isActiveTunnelServerEntry is used to check if there's already
  1227. // an existing tunnel to a candidate server.
  1228. func (controller *Controller) isActiveTunnelServerEntry(
  1229. serverEntry *protocol.ServerEntry) bool {
  1230. controller.tunnelMutex.Lock()
  1231. defer controller.tunnelMutex.Unlock()
  1232. for _, activeTunnel := range controller.tunnels {
  1233. if activeTunnel.dialParams.ServerEntry.IpAddress == serverEntry.IpAddress {
  1234. return true
  1235. }
  1236. }
  1237. return false
  1238. }
  1239. func (controller *Controller) setTunnelPoolSize(tunnelPoolSize int) {
  1240. controller.tunnelMutex.Lock()
  1241. defer controller.tunnelMutex.Unlock()
  1242. if tunnelPoolSize < 1 {
  1243. tunnelPoolSize = 1
  1244. }
  1245. if tunnelPoolSize > MAX_TUNNEL_POOL_SIZE {
  1246. tunnelPoolSize = MAX_TUNNEL_POOL_SIZE
  1247. }
  1248. controller.tunnelPoolSize = tunnelPoolSize
  1249. }
  1250. func (controller *Controller) getTunnelPoolSize() int {
  1251. controller.tunnelMutex.Lock()
  1252. defer controller.tunnelMutex.Unlock()
  1253. return controller.tunnelPoolSize
  1254. }
  1255. // Dial selects an active tunnel and establishes a port forward
  1256. // connection through the selected tunnel. Failure to connect is considered
  1257. // a port forward failure, for the purpose of monitoring tunnel health.
  1258. //
  1259. // When split tunnel mode is enabled, the connection may be untunneled,
  1260. // depending on GeoIP classification of the destination.
  1261. //
  1262. // downstreamConn is an optional parameter which specifies a connection to be
  1263. // explicitly closed when the dialed connection is closed. For instance, this
  1264. // is used to close downstreamConn App<->LocalProxy connections when the
  1265. // related LocalProxy<->SshPortForward connections close.
  1266. func (controller *Controller) Dial(
  1267. remoteAddr string, downstreamConn net.Conn) (conn net.Conn, err error) {
  1268. tunnel := controller.getNextActiveTunnel()
  1269. if tunnel == nil {
  1270. return nil, errors.TraceNew("no active tunnels")
  1271. }
  1272. if !tunnel.config.IsSplitTunnelEnabled() {
  1273. tunneledConn, splitTunnel, err := tunnel.DialTCPChannel(
  1274. remoteAddr, false, downstreamConn)
  1275. if err != nil {
  1276. return nil, errors.Trace(err)
  1277. }
  1278. if splitTunnel {
  1279. return nil, errors.TraceNew(
  1280. "unexpected split tunnel classification")
  1281. }
  1282. return tunneledConn, nil
  1283. }
  1284. // In split tunnel mode, TCP port forwards to destinations in the same
  1285. // country as the client are untunneled.
  1286. //
  1287. // Split tunnel is implemented with assistence from the server to classify
  1288. // destinations as being in the same country as the client. The server knows
  1289. // the client's public IP GeoIP data, and, for clients with split tunnel mode
  1290. // enabled, the server resolves the port forward destination address and
  1291. // checks the destination IP GeoIP data.
  1292. //
  1293. // When the countries match, the server "rejects" the port forward with a
  1294. // distinct response that indicates to the client that an untunneled port
  1295. // foward should be established locally.
  1296. //
  1297. // The client maintains a classification cache that allows it to make
  1298. // untunneled port forwards without requiring a round trip to the server.
  1299. // Only destinations classified as untunneled are stored in the cache: a
  1300. // destination classified as tunneled requires the same round trip as an
  1301. // unknown destination.
  1302. //
  1303. // When the countries do not match, the server establishes a port forward, as
  1304. // it does for all port forwards in non-split tunnel mode. There is no
  1305. // additional round trip for tunneled port forwards.
  1306. //
  1307. // Each destination includes a host and port. Since there are special
  1308. // cases where the server performs transparent redirection for specific
  1309. // host:port combinations, including UDPInterceptUdpgwServerAddress, the
  1310. // classification can differ for the same host but different ports and so
  1311. // the classification is cached using the full address, host:port, as the
  1312. // key. While this results in additional classification round trips for
  1313. // destinations with the same domain but differing ports, in practise
  1314. // most destinations use only port 443.
  1315. untunneledCache := controller.untunneledSplitTunnelClassifications
  1316. // If the destination is in the untunneled split tunnel classifications
  1317. // cache, skip the round trip to the server and do the direct, untunneled
  1318. // dial immediately.
  1319. _, cachedUntunneled := untunneledCache.Get(remoteAddr)
  1320. if !cachedUntunneled {
  1321. tunneledConn, splitTunnel, err := tunnel.DialTCPChannel(
  1322. remoteAddr, false, downstreamConn)
  1323. if err != nil {
  1324. return nil, errors.Trace(err)
  1325. }
  1326. if !splitTunnel {
  1327. // Clear any cached untunneled classification entry for this
  1328. // destination, as the server is now classifying it as tunneled.
  1329. untunneledCache.Delete(remoteAddr)
  1330. return tunneledConn, nil
  1331. }
  1332. // The server has indicated that the client should make a direct,
  1333. // untunneled dial. Cache the classification to avoid this round trip in
  1334. // the immediate future.
  1335. untunneledCache.Set(remoteAddr, true, lrucache.DefaultExpiration)
  1336. }
  1337. NoticeUntunneled(remoteAddr)
  1338. untunneledConn, err := controller.DirectDial(remoteAddr)
  1339. if err != nil {
  1340. return nil, errors.Trace(err)
  1341. }
  1342. return untunneledConn, nil
  1343. }
  1344. // DirectDial dials an untunneled TCP connection within the controller run context.
  1345. func (controller *Controller) DirectDial(remoteAddr string) (conn net.Conn, err error) {
  1346. return DialTCP(controller.runCtx, remoteAddr, controller.untunneledDialConfig)
  1347. }
  1348. // triggerFetches signals RSL, OSL, DSL, and upgrade download fetchers to
  1349. // begin, if not already running. triggerFetches is called when tunnel
  1350. // establishment fails to complete within a deadline and in other cases where
  1351. // local circumvention capabilities are lacking and we may require new server
  1352. // entries or client versions with new capabilities.
  1353. func (controller *Controller) triggerFetches() {
  1354. // Trigger a common remote server list fetch, since we may have failed
  1355. // to connect with all known servers. Don't block sending signal, since
  1356. // this signal may have already been sent.
  1357. // Don't wait for fetch remote to succeed, since it may fail and
  1358. // enter a retry loop and we're better off trying more known servers.
  1359. // TODO: synchronize the fetch response, so it can be incorporated
  1360. // into the server entry iterator as soon as available.
  1361. select {
  1362. case controller.signalFetchCommonRemoteServerList <- struct{}{}:
  1363. default:
  1364. }
  1365. // Trigger an OSL fetch in parallel. Server list fetches are run in
  1366. // parallel so that if one fetch is large or slow, it doesn't doesn't
  1367. // entirely block fetching the others.
  1368. select {
  1369. case controller.signalFetchObfuscatedServerLists <- struct{}{}:
  1370. default:
  1371. }
  1372. // Trigger the untunneled DSL fetch. The untunneled DSL fetch is similar
  1373. // to the classic RSL and OSL fetches in that it will attempt to download
  1374. // a larger, diverse selection of servers.
  1375. select {
  1376. case controller.signalUntunneledDSLFetch <- struct{}{}:
  1377. default:
  1378. }
  1379. // Trigger an out-of-band upgrade availability check and download.
  1380. // Since we may have failed to connect, we may benefit from upgrading
  1381. // to a new client version with new circumvention capabilities.
  1382. select {
  1383. case controller.signalDownloadUpgrade <- "":
  1384. default:
  1385. }
  1386. }
  1387. type protocolSelectionConstraints struct {
  1388. config *Config
  1389. initialLimitTunnelProtocols protocol.TunnelProtocols
  1390. initialLimitTunnelProtocolsCandidateCount int
  1391. limitTunnelProtocols protocol.TunnelProtocols
  1392. limitTunnelDialPortNumbers protocol.TunnelProtocolPortLists
  1393. limitQUICVersions protocol.QUICVersions
  1394. replayCandidateCount int
  1395. inproxyClientDialRateLimiter *rate.Limiter
  1396. }
  1397. func (p *protocolSelectionConstraints) hasInitialProtocols() bool {
  1398. return len(p.initialLimitTunnelProtocols) > 0 && p.initialLimitTunnelProtocolsCandidateCount > 0
  1399. }
  1400. func (p *protocolSelectionConstraints) isInitialCandidate(
  1401. excludeIntensive bool,
  1402. serverEntry *protocol.ServerEntry) bool {
  1403. return p.hasInitialProtocols() &&
  1404. len(serverEntry.GetSupportedProtocols(
  1405. conditionallyEnabledComponents{},
  1406. p.config.UseUpstreamProxy(),
  1407. p.initialLimitTunnelProtocols,
  1408. p.limitTunnelDialPortNumbers,
  1409. p.limitQUICVersions,
  1410. excludeIntensive)) > 0
  1411. }
  1412. func (p *protocolSelectionConstraints) isCandidate(
  1413. excludeIntensive bool,
  1414. serverEntry *protocol.ServerEntry) bool {
  1415. return len(serverEntry.GetSupportedProtocols(
  1416. conditionallyEnabledComponents{},
  1417. p.config.UseUpstreamProxy(),
  1418. p.limitTunnelProtocols,
  1419. p.limitTunnelDialPortNumbers,
  1420. p.limitQUICVersions,
  1421. excludeIntensive)) > 0
  1422. }
  1423. func (p *protocolSelectionConstraints) canReplay(
  1424. connectTunnelCount int,
  1425. excludeIntensive bool,
  1426. serverEntry *protocol.ServerEntry,
  1427. replayProtocol string) bool {
  1428. if p.replayCandidateCount != -1 && connectTunnelCount > p.replayCandidateCount {
  1429. return false
  1430. }
  1431. return common.Contains(
  1432. p.supportedProtocols(
  1433. connectTunnelCount, excludeIntensive, serverEntry),
  1434. replayProtocol)
  1435. }
  1436. func (p *protocolSelectionConstraints) getLimitTunnelProtocols(
  1437. connectTunnelCount int) protocol.TunnelProtocols {
  1438. protocols := p.limitTunnelProtocols
  1439. if len(p.initialLimitTunnelProtocols) > 0 &&
  1440. p.initialLimitTunnelProtocolsCandidateCount > connectTunnelCount {
  1441. protocols = p.initialLimitTunnelProtocols
  1442. }
  1443. return protocols
  1444. }
  1445. func (p *protocolSelectionConstraints) supportedProtocols(
  1446. connectTunnelCount int,
  1447. excludeIntensive bool,
  1448. serverEntry *protocol.ServerEntry) protocol.TunnelProtocols {
  1449. return serverEntry.GetSupportedProtocols(
  1450. conditionallyEnabledComponents{},
  1451. p.config.UseUpstreamProxy(),
  1452. p.getLimitTunnelProtocols(connectTunnelCount),
  1453. p.limitTunnelDialPortNumbers,
  1454. p.limitQUICVersions,
  1455. excludeIntensive)
  1456. }
  1457. func (p *protocolSelectionConstraints) selectProtocol(
  1458. connectTunnelCount int,
  1459. excludeIntensive bool,
  1460. preferInproxy bool,
  1461. serverEntry *protocol.ServerEntry) (string, time.Duration, bool) {
  1462. candidateProtocols := p.supportedProtocols(
  1463. connectTunnelCount, excludeIntensive, serverEntry)
  1464. // Prefer selecting an in-proxy tunnel protocol when indicated, but fall
  1465. // back to other protocols when no in-proxy protocol is supported.
  1466. if preferInproxy && candidateProtocols.HasInproxyTunnelProtocols() {
  1467. NoticeInfo("in-proxy protocol preferred")
  1468. candidateProtocols = candidateProtocols.PruneNonInproxyTunnelProtocols()
  1469. }
  1470. if len(candidateProtocols) == 0 {
  1471. return "", 0, false
  1472. }
  1473. // Pick at random from the supported protocols. This ensures that we'll
  1474. // eventually try all possible protocols. Depending on network
  1475. // configuration, it may be the case that some protocol is only available
  1476. // through multi-capability servers, and a simpler ranked preference of
  1477. // protocols could lead to that protocol never being selected.
  1478. selectedProtocol := candidateProtocols[prng.Intn(len(candidateProtocols))]
  1479. if !protocol.TunnelProtocolUsesInproxy(selectedProtocol) ||
  1480. p.inproxyClientDialRateLimiter == nil {
  1481. return selectedProtocol, 0, true
  1482. }
  1483. // Rate limit in-proxy dials. This avoids triggering rate limits or
  1484. // similar errors from any intermediate CDN between the client and the
  1485. // broker. And avoids unnecessarily triggering the broker's
  1486. // application-level rate limiter, which will incur some overhead logging
  1487. // an event and returning a response.
  1488. //
  1489. // In personal pairing mode, or when protocol limits yield only in-proxy
  1490. // tunnel protocol candidates, no non-in-proxy protocol can be selected,
  1491. // so delay the dial. In other cases, skip the candidate and pick a
  1492. // non-in-proxy tunnel protocol.
  1493. //
  1494. // Also delay, rather than skip, when preferring an in-proxy protocol.
  1495. // Note that in the prefer case, failure to meet requirements, such as
  1496. // having broker specs, will fail the dial and consume
  1497. // InproxyTunnelProtocolForceSelectionCount, when that mechanism is
  1498. // active. These fast failures should eventually lead to selecting
  1499. // non-in-proxy candidates; as a potential future enhancement, check the
  1500. // requirements _before_ applying InproxyTunnelProtocolPreferProbability
  1501. // or InproxyTunnelProtocolForceSelectionCount.
  1502. //
  1503. // The delay is not applied here since the caller is holding the
  1504. // concurrentEstablishTunnelsMutex lock, potentially blocking other
  1505. // establishment workers. Instead the delay is returned and applied
  1506. // outside of the lock. This also allows for the delay to be reduced when
  1507. // the StaggerConnectionWorkers facility is active.
  1508. //
  1509. // Limitation: potential fast dial failures may cause excess rate
  1510. // limiting, since tokens are consumed even if the dial fails before a
  1511. // request arrives at the broker. WaitForNetworkConnectivity, when
  1512. // configured, should pause calls to selectProtocol, although there are
  1513. // other possible fast fail cases.
  1514. //
  1515. // TODO: replace token on fast failure that doesn't reach the broker?
  1516. if preferInproxy ||
  1517. p.config.IsInproxyClientPersonalPairingMode() ||
  1518. p.getLimitTunnelProtocols(connectTunnelCount).IsOnlyInproxyTunnelProtocols() {
  1519. // Check for missing in-proxy broker request requirements before
  1520. // consuming a rate limit token.
  1521. //
  1522. // As a potential future enhancement, these checks, particularly
  1523. // haveInproxyCommonCompartmentIDs which reads and unmarshals a data
  1524. // store record, could be cached.
  1525. if !haveInproxyClientBrokerSpecs(p.config) {
  1526. NoticeInfo("in-proxy protocol selection failed: no broker specs")
  1527. return "", 0, false
  1528. }
  1529. if !p.config.IsInproxyClientPersonalPairingMode() &&
  1530. !haveInproxyCommonCompartmentIDs(p.config) {
  1531. NoticeInfo("in-proxy protocol selection failed: no common compartment IDs")
  1532. return "", 0, false
  1533. }
  1534. r := p.inproxyClientDialRateLimiter.Reserve()
  1535. if !r.OK() {
  1536. NoticeInfo("in-proxy protocol selection rate limited: burst size exceeded")
  1537. return "", 0, false
  1538. }
  1539. delay := r.Delay()
  1540. if delay > 0 {
  1541. NoticeInfo("in-proxy protocol selection rate limited: %v", delay)
  1542. }
  1543. return selectedProtocol, delay, true
  1544. } else {
  1545. // Check for missing in-proxy broker request requirements before
  1546. // consuming a rate limit token.
  1547. skip := true
  1548. if !haveInproxyClientBrokerSpecs(p.config) {
  1549. NoticeInfo("in-proxy protocol selection skipped: no broker specs")
  1550. } else if !haveInproxyCommonCompartmentIDs(p.config) {
  1551. NoticeInfo("in-proxy protocol selection skipped: no common compartment IDs")
  1552. } else if !p.inproxyClientDialRateLimiter.Allow() {
  1553. NoticeInfo("in-proxy protocol selection skipped: rate limit exceeded")
  1554. } else {
  1555. skip = false
  1556. }
  1557. if skip {
  1558. candidateProtocols = candidateProtocols.PruneInproxyTunnelProtocols()
  1559. if len(candidateProtocols) == 0 {
  1560. return "", 0, false
  1561. }
  1562. return candidateProtocols[prng.Intn(len(candidateProtocols))], 0, true
  1563. }
  1564. }
  1565. return selectedProtocol, 0, true
  1566. }
  1567. type candidateServerEntry struct {
  1568. serverEntry *protocol.ServerEntry
  1569. isServerAffinityCandidate bool
  1570. adjustedEstablishStartTime time.Time
  1571. }
  1572. // startEstablishing creates a pool of worker goroutines which will
  1573. // attempt to establish tunnels to candidate servers. The candidates
  1574. // are generated by another goroutine.
  1575. func (controller *Controller) startEstablishing() {
  1576. if controller.isEstablishing {
  1577. return
  1578. }
  1579. NoticeInfo("start establishing")
  1580. // establishStartTime is used to calculate and report the client's tunnel
  1581. // establishment duration. Establishment duration should include all
  1582. // initialization in launchEstablishing and establishCandidateGenerator,
  1583. // including any potentially long-running datastore iterations.
  1584. establishStartTime := time.Now()
  1585. controller.concurrentEstablishTunnelsMutex.Lock()
  1586. controller.establishConnectTunnelCount = 0
  1587. controller.concurrentEstablishTunnels = 0
  1588. controller.concurrentIntensiveEstablishTunnels = 0
  1589. controller.peakConcurrentEstablishTunnels = 0
  1590. controller.peakConcurrentIntensiveEstablishTunnels = 0
  1591. controller.establishInproxyForceSelectionCount = 0
  1592. controller.concurrentEstablishTunnelsMutex.Unlock()
  1593. DoGarbageCollection()
  1594. emitMemoryMetrics()
  1595. // The establish context cancelFunc, controller.stopEstablish, is called in
  1596. // controller.stopEstablishing.
  1597. controller.isEstablishing = true
  1598. controller.establishStartTime = establishStartTime
  1599. controller.establishCtx, controller.stopEstablish = context.WithCancel(controller.runCtx)
  1600. controller.establishWaitGroup = new(sync.WaitGroup)
  1601. controller.candidateServerEntries = make(chan *candidateServerEntry)
  1602. // The server affinity mechanism attempts to favor the previously
  1603. // used server when reconnecting. This is beneficial for user
  1604. // applications which expect consistency in user IP address (for
  1605. // example, a web site which prompts for additional user
  1606. // authentication when the IP address changes).
  1607. //
  1608. // Only the very first server, as determined by
  1609. // datastore.PromoteServerEntry(), is the server affinity candidate.
  1610. // Concurrent connections attempts to many servers are launched
  1611. // without delay, in case the affinity server connection fails.
  1612. // While the affinity server connection is outstanding, when any
  1613. // other connection is established, there is a short grace period
  1614. // delay before delivering the established tunnel; this allows some
  1615. // time for the affinity server connection to succeed first.
  1616. // When the affinity server connection fails, any other established
  1617. // tunnel is registered without delay.
  1618. //
  1619. // Note: the establishTunnelWorker that receives the affinity
  1620. // candidate is solely resonsible for closing
  1621. // controller.serverAffinityDoneBroadcast.
  1622. controller.serverAffinityDoneBroadcast = make(chan struct{})
  1623. // TODO: Add a buffer of 1 so we don't miss a signal while worker is
  1624. // starting? Trade-off is potential back-to-back fetches. As-is,
  1625. // establish will eventually signal another fetch.
  1626. controller.establishSignalForceTacticsFetch = make(chan struct{})
  1627. controller.establishWaitGroup.Add(1)
  1628. go controller.launchEstablishing()
  1629. }
  1630. func (controller *Controller) launchEstablishing() {
  1631. defer controller.establishWaitGroup.Done()
  1632. // Before starting the establish tunnel workers, get and apply tactics,
  1633. // launching a tactics request if required -- when there are no tactics,
  1634. // or the cached tactics have expired.
  1635. //
  1636. // Wait only TacticsWaitPeriod for the tactics request to complete (or
  1637. // fail) before proceeding with tunnel establishment, in case the tactics
  1638. // request is blocked or takes very long to complete.
  1639. //
  1640. // An in-flight tactics request uses meek in round tripper mode, which
  1641. // uses less resources than meek tunnel relay mode. For this reason, the
  1642. // tactics request is not counted in concurrentIntensiveEstablishTunnels.
  1643. //
  1644. // TODO: HTTP/2 uses significantly more memory, so perhaps
  1645. // concurrentIntensiveEstablishTunnels should be counted in that case.
  1646. //
  1647. // Any in-flight tactics request or pending retry will be
  1648. // canceled when establishment is stopped.
  1649. //
  1650. // In some cases, no tunnel establishment can succeed without a fresh
  1651. // tactics fetch, even if there is existing, non-expired cached tactics.
  1652. // Currently, cases include in-proxy personal pairing mode and limiting
  1653. // tunnel protocols to in-proxy, where broker specs are both required and
  1654. // obtained exclusively from tactics. It is possible that cached tactics
  1655. // are found and used, but broker configurations have recently changed
  1656. // away from the broker specs in cached tactics.
  1657. //
  1658. // Another scenario, with exclusively in-proxy tunnel protocols, is a
  1659. // fresh start with no embedded server entries, where the initial
  1660. // GetTactics will fail with "no capable servers".
  1661. //
  1662. // To handle these cases, when cached tactics are used or no tactics can
  1663. // be fetched, the tactics worker goroutine will remain running and await
  1664. // a signal to force a tactics fetch that ignores any stored/cached
  1665. // tactics. Multiple signals and fetch attempts are supported, to retry
  1666. // when a GetTactics fetch iteration fails, including the "no capable
  1667. // servers" case, which may only succeed after a concurrent server list
  1668. // fetch completes.
  1669. //
  1670. // Limitation: this mechanism doesn't force repeated tactics fetches after
  1671. // one success, which risks being excessive. There's at most one
  1672. // successful fetch per establishment run. As such, it remains remotely
  1673. // possible that a tactics change, such as new broker specs, deployed in
  1674. // the middle of an establishment run, won't be fetched. A user-initiated
  1675. // stop/start toggle will work around this.
  1676. if !controller.config.DisableTactics {
  1677. timeout := controller.config.GetParameters().Get().Duration(
  1678. parameters.TacticsWaitPeriod)
  1679. initialTacticsDone := make(chan struct{})
  1680. tacticsWaitPeriod := time.NewTimer(timeout)
  1681. defer tacticsWaitPeriod.Stop()
  1682. controller.establishWaitGroup.Add(1)
  1683. go func() {
  1684. defer controller.establishWaitGroup.Done()
  1685. useStoredTactics := true
  1686. fetched := GetTactics(
  1687. controller.establishCtx, controller.config, useStoredTactics)
  1688. close(initialTacticsDone)
  1689. if fetched {
  1690. return
  1691. }
  1692. for {
  1693. select {
  1694. case <-controller.establishCtx.Done():
  1695. return
  1696. case <-controller.establishSignalForceTacticsFetch:
  1697. }
  1698. useStoredTactics = false
  1699. fetched = GetTactics(
  1700. controller.establishCtx, controller.config, useStoredTactics)
  1701. if fetched {
  1702. // No more forced tactics fetches after the first success.
  1703. break
  1704. }
  1705. }
  1706. }()
  1707. select {
  1708. case <-initialTacticsDone:
  1709. case <-tacticsWaitPeriod.C:
  1710. }
  1711. tacticsWaitPeriod.Stop()
  1712. if controller.isStopEstablishing() {
  1713. // This check isn't strictly required but avoids the overhead of launching
  1714. // workers if establishment stopped while awaiting a tactics request.
  1715. return
  1716. }
  1717. }
  1718. // Initial- and LimitTunnelProtocols may be set by tactics.
  1719. //
  1720. // These protocol limits are fixed once per establishment, for
  1721. // consistent application of related probabilities (applied by
  1722. // ParametersAccessor.TunnelProtocols). The
  1723. // establishLimitTunnelProtocolsState field must be read-only after this
  1724. // point, allowing concurrent reads by establishment workers.
  1725. p := controller.config.GetParameters().Get()
  1726. controller.protocolSelectionConstraints = &protocolSelectionConstraints{
  1727. config: controller.config,
  1728. initialLimitTunnelProtocols: p.TunnelProtocols(parameters.InitialLimitTunnelProtocols),
  1729. initialLimitTunnelProtocolsCandidateCount: p.Int(parameters.InitialLimitTunnelProtocolsCandidateCount),
  1730. limitTunnelProtocols: p.TunnelProtocols(parameters.LimitTunnelProtocols),
  1731. limitTunnelDialPortNumbers: protocol.TunnelProtocolPortLists(
  1732. p.TunnelProtocolPortLists(parameters.LimitTunnelDialPortNumbers)),
  1733. replayCandidateCount: p.Int(parameters.ReplayCandidateCount),
  1734. inproxyClientDialRateLimiter: controller.inproxyClientDialRateLimiter,
  1735. }
  1736. // Adjust protocol limits for in-proxy personal proxy mode. In this mode,
  1737. // the client will make connections only through a proxy with the
  1738. // corresponding personal compartment ID, so non-in-proxy tunnel
  1739. // protocols are disabled.
  1740. if controller.config.IsInproxyClientPersonalPairingMode() {
  1741. if len(controller.protocolSelectionConstraints.initialLimitTunnelProtocols) > 0 {
  1742. controller.protocolSelectionConstraints.initialLimitTunnelProtocols =
  1743. controller.protocolSelectionConstraints.
  1744. initialLimitTunnelProtocols.PruneNonInproxyTunnelProtocols()
  1745. }
  1746. if len(controller.protocolSelectionConstraints.limitTunnelProtocols) > 0 {
  1747. controller.protocolSelectionConstraints.limitTunnelProtocols =
  1748. controller.protocolSelectionConstraints.
  1749. limitTunnelProtocols.PruneNonInproxyTunnelProtocols()
  1750. }
  1751. // This covers two cases: if there was no limitTunnelProtocols to
  1752. // start, then limit to any in-proxy tunnel protocol; or, if there
  1753. // was a limit but OnlyInproxyTunnelProtocols evaluates to an empty
  1754. // list, also set the limit to any in-proxy tunnel protocol.
  1755. if len(controller.protocolSelectionConstraints.limitTunnelProtocols) == 0 {
  1756. controller.protocolSelectionConstraints.limitTunnelProtocols =
  1757. protocol.InproxyTunnelProtocols
  1758. }
  1759. }
  1760. // Initialize the in-proxy client dial rate limiter, using the latest
  1761. // tactics. Rate limits are used in
  1762. // protocolSelectionConstraints.selectProtocol. When
  1763. // InproxyClientDialRateLimitQuantity is 0, there is no rate limit.
  1764. //
  1765. // The rate limiter is reset for each establishment, which ensures no
  1766. // delays carry over from a previous establishment run. However, this
  1767. // does mean that very frequent re-establishments may exceed the rate
  1768. // limit overall.
  1769. inproxyRateLimitQuantity := p.Int(parameters.InproxyClientDialRateLimitQuantity)
  1770. inproxyRateLimitInterval := p.Duration(parameters.InproxyClientDialRateLimitInterval)
  1771. if inproxyRateLimitQuantity > 0 {
  1772. controller.inproxyClientDialRateLimiter = rate.NewLimiter(
  1773. rate.Limit(float64(inproxyRateLimitQuantity)/inproxyRateLimitInterval.Seconds()),
  1774. inproxyRateLimitQuantity)
  1775. }
  1776. // InproxyTunnelProtocolForceSelectionCount forces the specified number of
  1777. // early candidates to select in-proxy protocols.
  1778. //
  1779. // Only server entries with INPROXY capabilities are counted as forced
  1780. // selection candidates; and, as currently implemented, these server
  1781. // entries are not sorted to the front of the server entry iterator, so
  1782. // force selection is applied opportunistically as server entries with
  1783. // the necessary capabilities are encountered.
  1784. //
  1785. // If a forced server entry has existing replay data for a non-in-proxy
  1786. // protocol, that replay data is ignored for this dial, but not deleted.
  1787. //
  1788. // The affinity server entry candidate is a potential candidate for forced
  1789. // selection.
  1790. controller.establishInproxyForceSelectionCount =
  1791. p.Int(parameters.InproxyTunnelProtocolForceSelectionCount)
  1792. // ConnectionWorkerPoolSize may be set by tactics.
  1793. //
  1794. // In-proxy personal pairing mode uses a distinct parameter which is
  1795. // typically configured to a lower number, limiting concurrent load and
  1796. // announcement consumption for personal proxies.
  1797. var workerPoolSize int
  1798. if controller.config.IsInproxyClientPersonalPairingMode() {
  1799. workerPoolSize = p.Int(parameters.InproxyPersonalPairingConnectionWorkerPoolSize)
  1800. } else {
  1801. workerPoolSize = p.Int(parameters.ConnectionWorkerPoolSize)
  1802. }
  1803. // When TargetServerEntry is used, override any worker pool size config or
  1804. // tactic parameter and use a pool size of 1. The typical use case for
  1805. // TargetServerEntry is to test a specific server with a single connection
  1806. // attempt. Furthermore, too many concurrent attempts to connect to the
  1807. // same server will trigger rate limiting.
  1808. if controller.config.TargetServerEntry != "" {
  1809. workerPoolSize = 1
  1810. }
  1811. // TunnelPoolSize may be set by tactics, subject to local constraints. A pool
  1812. // size of one is forced in packet tunnel mode or when using a
  1813. // TargetServerEntry. The tunnel pool size is reduced when there are
  1814. // insufficent known server entries, within the set region and protocol
  1815. // constraints, to satisfy the target.
  1816. //
  1817. // Limitations, to simplify concurrent access to shared state: a ceiling of
  1818. // MAX_TUNNEL_POOL_SIZE is enforced by setTunnelPoolSize; the tunnel pool
  1819. // size target is not re-adjusted after an API handshake, even though the
  1820. // handshake response may deliver new tactics, or prune server entries which
  1821. // were potential candidates; nor is the target re-adjusted after fetching
  1822. // new server entries during this establishment.
  1823. tunnelPoolSize := p.Int(parameters.TunnelPoolSize)
  1824. if controller.config.PacketTunnelTunFileDescriptor > 0 ||
  1825. controller.config.TargetServerEntry != "" {
  1826. tunnelPoolSize = 1
  1827. }
  1828. controller.setTunnelPoolSize(tunnelPoolSize)
  1829. p.Close()
  1830. // Trigger CandidateServers and AvailableEgressRegions notices. By default,
  1831. // this is an asynchronous operation, as the underlying full server entry
  1832. // list enumeration may be a slow operation.
  1833. //
  1834. // AvailableEgressRegions: after a fresh install, the outer client may not
  1835. // have a list of regions to display; and LimitTunnelProtocols may reduce the
  1836. // number of available regions.
  1837. //
  1838. // When the outer client receives NoticeAvailableEgressRegions and the
  1839. // configured EgressRegion is not included in the region list, the outer
  1840. // client _should_ stop tunnel-core and prompt the user to change the region
  1841. // selection, as there are insufficient servers/capabilities to establish a
  1842. // tunnel in the selected region.
  1843. //
  1844. // This report is delayed until after tactics are likely to be applied,
  1845. // above; this avoids a ReportAvailableRegions reporting too many regions,
  1846. // followed shortly by a ReportAvailableRegions reporting fewer regions. That
  1847. // sequence could cause issues in the outer client UI.
  1848. //
  1849. // The reported regions are limited by protocolSelectionConstraints; in the
  1850. // case where an initial limit is in place, only regions available for the
  1851. // initial limit are reported. The initial phase will not complete if
  1852. // EgressRegion is set such that there are no server entries with the
  1853. // necessary protocol capabilities (either locally or from a remote server
  1854. // list fetch).
  1855. // Concurrency note: controller.protocolSelectionConstraints and its
  1856. // fields may be overwritten before serverEntriesReporter reads it, and
  1857. // so cannot be accessed directly by serverEntriesReporter.
  1858. //
  1859. // Limitation: the non-deep copy here shares slices (tunnel protocol
  1860. // lists) with the original; the contents of these slices don't change
  1861. // past this point. The rate limiter should not be used by
  1862. // serverEntriesReporter, but is cleared just in case.
  1863. copyConstraints := *controller.protocolSelectionConstraints
  1864. copyConstraints.inproxyClientDialRateLimiter = nil
  1865. controller.signalServerEntriesReporter(
  1866. &serverEntriesReportRequest{
  1867. constraints: &copyConstraints,
  1868. })
  1869. if controller.protocolSelectionConstraints.hasInitialProtocols() ||
  1870. tunnelPoolSize > 1 {
  1871. // Perform a synchronous scan over server entries in order to check if
  1872. // there are sufficient candidates to satisfy any initial tunnel
  1873. // protocol limit constraint and/or tunnel pool size > 1. If these
  1874. // requirements can't be met, the constraint and/or pool size are
  1875. // adjusted in order to avoid spinning unable to select any protocol
  1876. // or trying to establish more tunnels than is possible.
  1877. controller.doConstraintsScan()
  1878. }
  1879. for i := 0; i < workerPoolSize; i++ {
  1880. controller.establishWaitGroup.Add(1)
  1881. go controller.establishTunnelWorker()
  1882. }
  1883. controller.establishWaitGroup.Add(1)
  1884. go controller.establishCandidateGenerator()
  1885. }
  1886. func (controller *Controller) doConstraintsScan() {
  1887. // Scan over server entries in order to check and adjust any initial
  1888. // tunnel protocol limit and tunnel pool size.
  1889. //
  1890. // The scan in serverEntriesReporter is _not_ used for these checks,
  1891. // since it takes too long to complete with 1000s of server entries,
  1892. // greatly delaying the start(or restart, if already scanning) of
  1893. // establishment. Instead a 2nd ScanServerEntries is run here, with an
  1894. // early exit when sufficient candidates are found, which is expected
  1895. // to happen quickly in the typical case.
  1896. hasInitialLimitTunnelProtocols :=
  1897. controller.protocolSelectionConstraints.hasInitialProtocols()
  1898. tunnelPoolSize := controller.getTunnelPoolSize()
  1899. scanCount := 0
  1900. scanCancelled := false
  1901. candidates := 0
  1902. callback := func(serverEntry *protocol.ServerEntry) bool {
  1903. scanCount += 1
  1904. // As in serverEntryReporter:
  1905. // - egress region is ignored, since AvailableEgressRegion alerts
  1906. // the front end client when unable to connect due to egress
  1907. // region constraints.
  1908. // - excludeIntensive is false, as any intensive candidate will
  1909. // eventually be an available candidate.
  1910. excludeIntensive := false
  1911. if (hasInitialLimitTunnelProtocols &&
  1912. controller.protocolSelectionConstraints.isInitialCandidate(excludeIntensive, serverEntry)) ||
  1913. (!hasInitialLimitTunnelProtocols &&
  1914. controller.protocolSelectionConstraints.isCandidate(excludeIntensive, serverEntry)) {
  1915. candidates += 1
  1916. }
  1917. if candidates >= tunnelPoolSize {
  1918. // Exit the scan early once sufficient candidates have been found.
  1919. scanCancelled = true
  1920. return false
  1921. }
  1922. select {
  1923. case <-controller.runCtx.Done():
  1924. // Don't block controller shutdown: cancel the scan.
  1925. return false
  1926. default:
  1927. return true
  1928. }
  1929. }
  1930. startTime := time.Now()
  1931. scanErr := ScanServerEntries(callback)
  1932. if scanErr != nil && !scanCancelled {
  1933. NoticeWarning("ScanServerEntries failed: %v", errors.Trace(scanErr))
  1934. // Continue and make adjustments based on any partial results.
  1935. }
  1936. NoticeInfo("Awaited ScanServerEntries: scanned %d entries in %v", scanCount, time.Since(startTime))
  1937. // Make adjustments based on candidate counts.
  1938. if tunnelPoolSize > candidates && candidates > 0 {
  1939. tunnelPoolSize = candidates
  1940. }
  1941. // If InitialLimitTunnelProtocols is configured but cannot be satisfied,
  1942. // skip the initial phase in this establishment. This avoids spinning,
  1943. // unable to connect, in this case. InitialLimitTunnelProtocols is
  1944. // intended to prioritize certain protocols, but not strictly select them.
  1945. //
  1946. // The candidate count check ignores egress region selection. When an egress
  1947. // region is selected, it's the responsibility of the outer client to react
  1948. // to the following ReportAvailableRegions output and clear the user's
  1949. // selected region to prevent spinning, unable to connect. The initial phase
  1950. // is skipped only when InitialLimitTunnelProtocols cannot be satisfied
  1951. // _regardless_ of region selection.
  1952. //
  1953. // We presume that, in practise, most clients will have embedded server
  1954. // entries with capabilities for most protocols; and that clients will
  1955. // often perform RSL checks. So clients should most often have the
  1956. // necessary capabilities to satisfy InitialLimitTunnelProtocols. When
  1957. // this check fails, RSL/OSL/upgrade checks are triggered in order to gain
  1958. // new capabilities.
  1959. //
  1960. // LimitTunnelProtocols remains a hard limit, as using prohibited
  1961. // protocols may have some bad effect, such as a firewall blocking all
  1962. // traffic from a host.
  1963. if hasInitialLimitTunnelProtocols && candidates == 0 {
  1964. NoticeWarning("skipping initial limit tunnel protocols")
  1965. controller.protocolSelectionConstraints.initialLimitTunnelProtocolsCandidateCount = 0
  1966. // Since we were unable to satisfy the InitialLimitTunnelProtocols
  1967. // tactic, trigger RSL, OSL, and upgrade fetches to potentially
  1968. // gain new capabilities.
  1969. controller.triggerFetches()
  1970. }
  1971. }
  1972. // stopEstablishing signals the establish goroutines to stop and waits
  1973. // for the group to halt.
  1974. func (controller *Controller) stopEstablishing() {
  1975. if !controller.isEstablishing {
  1976. return
  1977. }
  1978. NoticeInfo("stop establishing")
  1979. controller.stopEstablish()
  1980. // Note: establishCandidateGenerator closes controller.candidateServerEntries
  1981. // (as it may be sending to that channel).
  1982. controller.establishWaitGroup.Wait()
  1983. NoticeInfo("stopped establishing")
  1984. controller.isEstablishing = false
  1985. controller.establishStartTime = time.Time{}
  1986. controller.establishCtx = nil
  1987. controller.stopEstablish = nil
  1988. controller.establishWaitGroup = nil
  1989. controller.candidateServerEntries = nil
  1990. controller.serverAffinityDoneBroadcast = nil
  1991. controller.establishSignalForceTacticsFetch = nil
  1992. controller.inproxyClientDialRateLimiter = nil
  1993. controller.concurrentEstablishTunnelsMutex.Lock()
  1994. peakConcurrent := controller.peakConcurrentEstablishTunnels
  1995. peakConcurrentIntensive := controller.peakConcurrentIntensiveEstablishTunnels
  1996. controller.establishConnectTunnelCount = 0
  1997. controller.concurrentEstablishTunnels = 0
  1998. controller.concurrentIntensiveEstablishTunnels = 0
  1999. controller.peakConcurrentEstablishTunnels = 0
  2000. controller.peakConcurrentIntensiveEstablishTunnels = 0
  2001. controller.concurrentEstablishTunnelsMutex.Unlock()
  2002. NoticeInfo("peak concurrent establish tunnels: %d", peakConcurrent)
  2003. NoticeInfo("peak concurrent resource intensive establish tunnels: %d", peakConcurrentIntensive)
  2004. emitMemoryMetrics()
  2005. DoGarbageCollection()
  2006. // Record datastore metrics after establishment, the phase which generates
  2007. // the bulk of all datastore transactions: iterating over server entries,
  2008. // storing new server entries, etc.
  2009. emitDatastoreMetrics()
  2010. // Similarly, establishment generates the bulk of domain resolves.
  2011. emitDNSMetrics(controller.resolver)
  2012. }
  2013. // establishCandidateGenerator populates the candidate queue with server entries
  2014. // from the data store. Server entries are iterated in rank order, so that promoted
  2015. // servers with higher rank are priority candidates.
  2016. func (controller *Controller) establishCandidateGenerator() {
  2017. defer controller.establishWaitGroup.Done()
  2018. defer close(controller.candidateServerEntries)
  2019. // networkWaitDuration is the elapsed time spent waiting
  2020. // for network connectivity. This duration will be excluded
  2021. // from reported tunnel establishment duration.
  2022. var totalNetworkWaitDuration time.Duration
  2023. applyServerAffinity, iterator, err := NewServerEntryIterator(controller.config)
  2024. if err != nil {
  2025. NoticeError("failed to iterate over candidates: %v", errors.Trace(err))
  2026. controller.SignalComponentFailure()
  2027. return
  2028. }
  2029. defer iterator.Close()
  2030. // TODO: reconcile server affinity scheme with multi-tunnel mode
  2031. if controller.getTunnelPoolSize() > 1 {
  2032. applyServerAffinity = false
  2033. }
  2034. isServerAffinityCandidate := true
  2035. if !applyServerAffinity {
  2036. isServerAffinityCandidate = false
  2037. close(controller.serverAffinityDoneBroadcast)
  2038. }
  2039. loop:
  2040. // Repeat until stopped
  2041. for {
  2042. // A "round" consists of a new shuffle of the server entries and attempted
  2043. // connections up to the end of the server entry iterator, or
  2044. // parameters.EstablishTunnelWorkTime elapsed. Time spent waiting for
  2045. // network connectivity is excluded from round elapsed time.
  2046. //
  2047. // After a round, if parameters.EstablishTunnelWorkTime has elapsed in total
  2048. // with no tunnel established, remote server list and upgrade checks are
  2049. // triggered.
  2050. //
  2051. // A complete server entry iteration does not trigger fetches since it's
  2052. // possible to have fewer than parameters.ConnectionWorkerPoolSize
  2053. // candidates, in which case rounds end instantly due to the complete server
  2054. // entry iteration. An exception is made for an empty server entry iterator;
  2055. // in that case fetches may be triggered immediately.
  2056. //
  2057. // The number of server candidates may change during this loop, due to
  2058. // remote server list fetches. Due to the performance impact, we will not
  2059. // trigger additional, informational CandidateServer notices while in the
  2060. // establishing loop. Clients typically re-establish often enough that we
  2061. // will see the effect of the remote server list fetch in diagnostics.
  2062. roundStartTime := time.Now()
  2063. var roundNetworkWaitDuration time.Duration
  2064. workTime := controller.config.GetParameters().Get().Duration(
  2065. parameters.EstablishTunnelWorkTime)
  2066. candidateServerEntryCount := 0
  2067. // Send each iterator server entry to the establish workers
  2068. for {
  2069. networkWaitStartTime := time.Now()
  2070. if !WaitForNetworkConnectivity(
  2071. controller.establishCtx,
  2072. controller.config.NetworkConnectivityChecker,
  2073. nil) {
  2074. break loop
  2075. }
  2076. networkWaitDuration := time.Since(networkWaitStartTime)
  2077. roundNetworkWaitDuration += networkWaitDuration
  2078. totalNetworkWaitDuration += networkWaitDuration
  2079. serverEntry, err := iterator.Next()
  2080. if err != nil {
  2081. NoticeError("failed to get next candidate: %v", errors.Trace(err))
  2082. controller.SignalComponentFailure()
  2083. return
  2084. }
  2085. if serverEntry == nil {
  2086. // Completed this iteration
  2087. NoticeInfo("completed server entry iteration")
  2088. break
  2089. }
  2090. if controller.config.TargetAPIProtocol == protocol.PSIPHON_API_PROTOCOL_SSH &&
  2091. !serverEntry.SupportsSSHAPIRequests() {
  2092. continue
  2093. }
  2094. candidateServerEntryCount += 1
  2095. // adjustedEstablishStartTime is establishStartTime shifted
  2096. // to exclude time spent waiting for network connectivity.
  2097. adjustedEstablishStartTime := controller.establishStartTime.Add(
  2098. totalNetworkWaitDuration)
  2099. candidate := &candidateServerEntry{
  2100. serverEntry: serverEntry,
  2101. isServerAffinityCandidate: isServerAffinityCandidate,
  2102. adjustedEstablishStartTime: adjustedEstablishStartTime,
  2103. }
  2104. wasServerAffinityCandidate := isServerAffinityCandidate
  2105. // Note: there must be only one server affinity candidate, as it
  2106. // closes the serverAffinityDoneBroadcast channel.
  2107. isServerAffinityCandidate = false
  2108. // TODO: here we could generate multiple candidates from the
  2109. // server entry when there are many MeekFrontingAddresses.
  2110. select {
  2111. case controller.candidateServerEntries <- candidate:
  2112. case <-controller.establishCtx.Done():
  2113. break loop
  2114. }
  2115. if time.Since(roundStartTime)-roundNetworkWaitDuration > workTime {
  2116. // Start over, after a brief pause, with a new shuffle of the server
  2117. // entries, and potentially some newly fetched server entries.
  2118. break
  2119. }
  2120. if wasServerAffinityCandidate {
  2121. // Don't start the next candidate until either the server affinity
  2122. // candidate has completed (success or failure) or is still working
  2123. // and the grace period has elapsed.
  2124. gracePeriod := controller.config.GetParameters().Get().Duration(
  2125. parameters.EstablishTunnelServerAffinityGracePeriod)
  2126. if gracePeriod > 0 {
  2127. timer := time.NewTimer(gracePeriod)
  2128. select {
  2129. case <-timer.C:
  2130. case <-controller.serverAffinityDoneBroadcast:
  2131. case <-controller.establishCtx.Done():
  2132. timer.Stop()
  2133. break loop
  2134. }
  2135. timer.Stop()
  2136. }
  2137. }
  2138. }
  2139. // Free up resources now, but don't reset until after the pause.
  2140. iterator.Close()
  2141. // Trigger RSL, OSL, and upgrade checks after failing to establish a
  2142. // tunnel within parameters.EstablishTunnelWorkTime, or if there are
  2143. // no server entries present.
  2144. //
  2145. // While the trigger is made after each round,
  2146. // parameter.FetchRemoteServerListStalePeriod will limit the actual
  2147. // frequency of fetches. Continuing to trigger allows for very long running
  2148. // establishments to perhaps eventually succeed.
  2149. //
  2150. // No fetches are triggered when TargetServerEntry is specified. In that
  2151. // case, we're only trying to connect to a specific server entry.
  2152. if candidateServerEntryCount == 0 ||
  2153. time.Since(controller.establishStartTime)-totalNetworkWaitDuration > workTime {
  2154. if controller.config.TargetServerEntry == "" {
  2155. controller.triggerFetches()
  2156. }
  2157. // Trigger a forced tactics fetch. Currently, this is done only
  2158. // for cases where in-proxy tunnel protocols must be selected.
  2159. // When there were no server entries, wait until a server entry
  2160. // fetch has completed.
  2161. // Lock required to access controller.establishConnectTunnelCount.
  2162. controller.concurrentEstablishTunnelsMutex.Lock()
  2163. limitInproxyOnly := controller.protocolSelectionConstraints.getLimitTunnelProtocols(
  2164. controller.establishConnectTunnelCount).IsOnlyInproxyTunnelProtocols()
  2165. controller.concurrentEstablishTunnelsMutex.Unlock()
  2166. if limitInproxyOnly || controller.config.IsInproxyClientPersonalPairingMode() {
  2167. // Simply sleep and poll for any imported server entries;
  2168. // perform one sleep after HasServerEntries, in order to give
  2169. // the import some extra time. Limitation: if the sleep loop
  2170. // ends too soon, the tactics fetch won't find a
  2171. // tactics-capable server entry; in this case, workTime must
  2172. // elapse before another tactics fetch is triggered.
  2173. //
  2174. // TODO: synchronize with server list fetch/import complete;
  2175. // or use ScanServerEntries (but see function comment about
  2176. // performance concern) to check for at least one
  2177. // tactics-capable server entry.
  2178. if candidateServerEntryCount == 0 {
  2179. stopWaiting := false
  2180. for {
  2181. if HasServerEntries() {
  2182. stopWaiting = true
  2183. }
  2184. common.SleepWithContext(controller.establishCtx, 1*time.Second)
  2185. if stopWaiting || controller.establishCtx.Err() != nil {
  2186. break
  2187. }
  2188. }
  2189. }
  2190. select {
  2191. case controller.establishSignalForceTacticsFetch <- struct{}{}:
  2192. default:
  2193. }
  2194. }
  2195. }
  2196. // After a complete iteration of candidate servers, pause before iterating again.
  2197. // This helps avoid some busy wait loop conditions, and also allows some time for
  2198. // network conditions to change. Also allows for fetch remote to complete,
  2199. // in typical conditions (it isn't strictly necessary to wait for this, there will
  2200. // be more rounds if required).
  2201. p := controller.config.GetParameters().Get()
  2202. timeout := prng.JitterDuration(
  2203. p.Duration(parameters.EstablishTunnelPausePeriod),
  2204. p.Float(parameters.EstablishTunnelPausePeriodJitter))
  2205. p.Close()
  2206. timer := time.NewTimer(timeout)
  2207. select {
  2208. case <-timer.C:
  2209. // Retry iterating
  2210. case <-controller.establishCtx.Done():
  2211. timer.Stop()
  2212. break loop
  2213. }
  2214. timer.Stop()
  2215. err := iterator.Reset()
  2216. if err != nil {
  2217. NoticeError("failed to reset iterator: %v", errors.Trace(err))
  2218. controller.SignalComponentFailure()
  2219. return
  2220. }
  2221. }
  2222. }
  2223. // establishTunnelWorker pulls candidates from the candidate queue, establishes
  2224. // a connection to the tunnel server, and delivers the connected tunnel to a channel.
  2225. func (controller *Controller) establishTunnelWorker() {
  2226. defer controller.establishWaitGroup.Done()
  2227. loop:
  2228. for candidateServerEntry := range controller.candidateServerEntries {
  2229. // Note: don't receive from candidateServerEntries and isStopEstablishing
  2230. // in the same select, since we want to prioritize receiving the stop signal
  2231. if controller.isStopEstablishing() {
  2232. break loop
  2233. }
  2234. // There may already be a tunnel to this candidate. If so, skip it.
  2235. if controller.isActiveTunnelServerEntry(candidateServerEntry.serverEntry) {
  2236. continue
  2237. }
  2238. // TODO: we allow multiple, concurrent workers to attempt to connect to the
  2239. // same server. This is not wasteful if the server supports several
  2240. // different protocols, some of which may be blocked while others are not
  2241. // blocked. Limiting protocols with [Initial]LimitTunnelProtocols may make
  2242. // these multiple attempts redundent. Also, replay should be used only by
  2243. // the first attempt.
  2244. // upstreamProxyErrorCallback will post NoticeUpstreamProxyError when the
  2245. // tunnel dial fails due to an upstream proxy error. As the upstream proxy
  2246. // is user configured, the error message may need to be relayed to the user.
  2247. // As the callback may be invoked after establishment is over (e.g., if an
  2248. // initial dial isn't fully shutdown when ConnectTunnel returns; or a meek
  2249. // underlying TCP connection re-dial) don't access these variables
  2250. // directly.
  2251. callbackCandidateServerEntry := candidateServerEntry
  2252. callbackEstablishCtx := controller.establishCtx
  2253. upstreamProxyErrorCallback := func(err error) {
  2254. // Do not post the notice when overall establishment context is canceled or
  2255. // timed-out: the upstream proxy connection error is likely a result of the
  2256. // cancellation, and not a condition to be fixed by the user. In the case
  2257. // of meek underlying TCP connection re-dials, this condition will always
  2258. // be true; however in this case the initial dial succeeded with the
  2259. // current upstream proxy settings, so any upstream proxy error is
  2260. // transient.
  2261. if callbackEstablishCtx.Err() != nil {
  2262. return
  2263. }
  2264. // Another class of non-fatal upstream proxy error arises from proxies
  2265. // which limit permitted proxied ports. In this case, some tunnels may fail
  2266. // due to dial port, while others may eventually succeed. To avoid this
  2267. // class of errors, delay posting the notice. If the upstream proxy works,
  2268. // _some_ tunnel should connect. If the upstream proxy configuration is
  2269. // broken, the error should persist and eventually get posted.
  2270. p := controller.config.GetParameters().Get()
  2271. workerPoolSize := p.Int(parameters.ConnectionWorkerPoolSize)
  2272. minWaitDuration := p.Duration(parameters.UpstreamProxyErrorMinWaitDuration)
  2273. maxWaitDuration := p.Duration(parameters.UpstreamProxyErrorMaxWaitDuration)
  2274. p.Close()
  2275. controller.concurrentEstablishTunnelsMutex.Lock()
  2276. establishConnectTunnelCount := controller.establishConnectTunnelCount
  2277. controller.concurrentEstablishTunnelsMutex.Unlock()
  2278. // Delay UpstreamProxyErrorMinWaitDuration (excluding time spent waiting
  2279. // for network connectivity) and then until either
  2280. // UpstreamProxyErrorMaxWaitDuration has elapsed or, to post sooner if many
  2281. // candidates are failing, at least workerPoolSize tunnel connection
  2282. // attempts have completed. We infer that at least workerPoolSize
  2283. // candidates have completed by checking that at least 2*workerPoolSize
  2284. // candidates have started.
  2285. elapsedTime := time.Since(
  2286. callbackCandidateServerEntry.adjustedEstablishStartTime)
  2287. if elapsedTime < minWaitDuration ||
  2288. (elapsedTime < maxWaitDuration &&
  2289. establishConnectTunnelCount < 2*workerPoolSize) {
  2290. return
  2291. }
  2292. NoticeUpstreamProxyError(err)
  2293. }
  2294. // Select the tunnel protocol. The selection will be made at random
  2295. // from protocols supported by the server entry, optionally limited by
  2296. // LimitTunnelProtocols.
  2297. //
  2298. // When limiting concurrent resource intensive protocol connection
  2299. // workers, and at the limit, do not select resource intensive
  2300. // protocols since otherwise the candidate must be skipped.
  2301. //
  2302. // If at the limit and unabled to select a non-intensive protocol,
  2303. // skip the candidate entirely and move on to the next. Since
  2304. // candidates are shuffled it's likely that the next candidate is not
  2305. // intensive. In this case, a StaggerConnectionWorkersMilliseconds
  2306. // delay may still be incurred.
  2307. p := controller.config.GetParameters().Get()
  2308. limitIntensiveConnectionWorkers := p.Int(parameters.LimitIntensiveConnectionWorkers)
  2309. inproxyPreferProbability := p.Float(parameters.InproxyTunnelProtocolPreferProbability)
  2310. staggerPeriod := p.Duration(parameters.StaggerConnectionWorkersPeriod)
  2311. staggerJitter := p.Float(parameters.StaggerConnectionWorkersJitter)
  2312. p.Close()
  2313. // Access to controller fields is synchronized with this lock. The
  2314. // canReplay and selectProtocol callbacks are intended to be invoked
  2315. // in MakeDialParameters while lock is held.
  2316. controller.concurrentEstablishTunnelsMutex.Lock()
  2317. excludeIntensive := false
  2318. if limitIntensiveConnectionWorkers > 0 &&
  2319. controller.concurrentIntensiveEstablishTunnels >= limitIntensiveConnectionWorkers {
  2320. excludeIntensive = true
  2321. }
  2322. // Force in-proxy protocol selection as required, and if the server
  2323. // entry supports in-proxy protocols. If this candidate happens to be
  2324. // a replay of an in-proxy protocol, it's still counted as a forced
  2325. // selection.
  2326. //
  2327. // Forced selection is skipped when excluding intensive protocols, as
  2328. // TunnelProtocolIsResourceIntensive currently includes
  2329. // TunnelProtocolUsesInproxy.
  2330. inproxyForceSelection := false
  2331. if !excludeIntensive &&
  2332. controller.establishInproxyForceSelectionCount > 0 &&
  2333. controller.protocolSelectionConstraints.supportedProtocols(
  2334. controller.establishConnectTunnelCount,
  2335. excludeIntensive,
  2336. candidateServerEntry.serverEntry).HasInproxyTunnelProtocols() {
  2337. NoticeInfo("in-proxy protocol selection forced")
  2338. inproxyForceSelection = true
  2339. controller.establishInproxyForceSelectionCount -= 1
  2340. }
  2341. canReplay := func(serverEntry *protocol.ServerEntry, replayProtocol string) bool {
  2342. if inproxyForceSelection {
  2343. if !protocol.TunnelProtocolUsesInproxy(replayProtocol) {
  2344. // Skip replay when forcing in-proxy protocol selection.
  2345. // MakeDialParameters will call the following
  2346. // selectProtocol callback with in-proxy preferred.
  2347. //
  2348. // Skipping here retains the existing replay data, as
  2349. // DialParameters.Failed will only delete it when
  2350. // IsReplay. However, the old replay data can be replaced
  2351. // if the in-proxy tunnel is successful.
  2352. return false
  2353. } else {
  2354. // MakeDialParameters calls canReplay only once it has
  2355. // replay data for the server entry candidate, so this
  2356. // will be a replay.
  2357. NoticeInfo("in-proxy protocol selection replayed")
  2358. return true
  2359. }
  2360. }
  2361. return controller.protocolSelectionConstraints.canReplay(
  2362. controller.establishConnectTunnelCount,
  2363. excludeIntensive,
  2364. serverEntry,
  2365. replayProtocol)
  2366. }
  2367. // The dial rate limit delay, determined by protocolSelectionConstraints.selectProtocol, is
  2368. // not applied within that function since this worker holds the concurrentEstablishTunnelsMutex
  2369. // lock when that's called. Instead, the required delay is passed out and applied below.
  2370. // It's safe for the selectProtocol callback to write to dialRateLimitDelay without
  2371. // synchronization since this worker goroutine invokes the callback.
  2372. var dialRateLimitDelay time.Duration
  2373. selectProtocol := func(
  2374. serverEntry *protocol.ServerEntry) (string, bool) {
  2375. preferInproxy := inproxyForceSelection || prng.FlipWeightedCoin(inproxyPreferProbability)
  2376. selectedProtocol, rateLimitDelay, ok := controller.protocolSelectionConstraints.selectProtocol(
  2377. controller.establishConnectTunnelCount,
  2378. excludeIntensive,
  2379. preferInproxy,
  2380. serverEntry)
  2381. dialRateLimitDelay = rateLimitDelay
  2382. return selectedProtocol, ok
  2383. }
  2384. // MakeDialParameters may return a replay instance, if the server
  2385. // entry has a previous, recent successful connection and
  2386. // tactics/config has not changed.
  2387. //
  2388. // In the first round -- and later rounds, with some probability -- of
  2389. // establishing, ServerEntryIterator will move potential replay candidates
  2390. // to the front of the iterator after the random shuffle, which greatly
  2391. // prioritizes previously successful servers for that round.
  2392. //
  2393. // As ServerEntryIterator does not unmarshal and validate replay
  2394. // candidate dial parameters, some potential replay candidates may
  2395. // have expired or otherwise ineligible dial parameters; in this case
  2396. // the candidate proceeds without replay.
  2397. //
  2398. // The ReplayCandidateCount tactic determines how many candidates may use
  2399. // replay. After ReplayCandidateCount candidates of any type, replay or no,
  2400. // replay is skipped. If ReplayCandidateCount exceeds the intial round,
  2401. // replay may still be performed but the iterator may no longer move
  2402. // potential replay server entries to the front. When ReplayCandidateCount
  2403. // is set to -1, unlimited candidates may use replay.
  2404. dialParams, err := MakeDialParameters(
  2405. controller.config,
  2406. controller.steeringIPCache,
  2407. controller.quicTLSClientSessionCache,
  2408. controller.tlsClientSessionCache,
  2409. upstreamProxyErrorCallback,
  2410. canReplay,
  2411. selectProtocol,
  2412. candidateServerEntry.serverEntry,
  2413. controller.inproxyClientBrokerClientManager,
  2414. controller.inproxyNATStateManager,
  2415. false,
  2416. controller.establishConnectTunnelCount,
  2417. int(atomic.LoadInt32(&controller.establishedTunnelsCount)))
  2418. if dialParams == nil || err != nil {
  2419. controller.concurrentEstablishTunnelsMutex.Unlock()
  2420. // MakeDialParameters returns nil/nil when the server entry is to
  2421. // be skipped. See MakeDialParameters for skip cases and skip
  2422. // logging. Silently fail the candidate in this case. Otherwise,
  2423. // emit error.
  2424. if err != nil {
  2425. NoticeInfo("failed to make dial parameters for %s: %v",
  2426. candidateServerEntry.serverEntry.GetDiagnosticID(),
  2427. errors.Trace(err))
  2428. }
  2429. // Unblock other candidates immediately when server affinity
  2430. // candidate is skipped.
  2431. if candidateServerEntry.isServerAffinityCandidate {
  2432. close(controller.serverAffinityDoneBroadcast)
  2433. }
  2434. continue
  2435. }
  2436. // Increment establishConnectTunnelCount only after selectProtocol has
  2437. // succeeded to ensure InitialLimitTunnelProtocolsCandidateCount
  2438. // candidates use InitialLimitTunnelProtocols.
  2439. //
  2440. // TODO: add escape from initial limit to cover cases where the
  2441. // initial scan indicates there are sufficient candidates, but then
  2442. // server entries are deleted.
  2443. establishConnectTunnelCount := controller.establishConnectTunnelCount
  2444. controller.establishConnectTunnelCount += 1
  2445. isIntensive := protocol.TunnelProtocolIsResourceIntensive(dialParams.TunnelProtocol)
  2446. if isIntensive {
  2447. controller.concurrentIntensiveEstablishTunnels += 1
  2448. if controller.concurrentIntensiveEstablishTunnels > controller.peakConcurrentIntensiveEstablishTunnels {
  2449. controller.peakConcurrentIntensiveEstablishTunnels = controller.concurrentIntensiveEstablishTunnels
  2450. }
  2451. }
  2452. controller.concurrentEstablishTunnels += 1
  2453. if controller.concurrentEstablishTunnels > controller.peakConcurrentEstablishTunnels {
  2454. controller.peakConcurrentEstablishTunnels = controller.concurrentEstablishTunnels
  2455. }
  2456. controller.concurrentEstablishTunnelsMutex.Unlock()
  2457. startStagger := time.Now()
  2458. // Apply stagger only now that we're past MakeDialParameters and
  2459. // protocol selection logic which may have caused the candidate to be
  2460. // skipped. The stagger logic delays dialing, and we don't want to
  2461. // incur that delay that when skipping.
  2462. //
  2463. // Locking staggerMutex serializes staggers, so that multiple workers
  2464. // don't simply sleep in parallel.
  2465. //
  2466. // The stagger is applied when establishConnectTunnelCount > 0 -- that
  2467. // is, for all but the first dial.
  2468. if establishConnectTunnelCount > 0 && staggerPeriod != 0 {
  2469. controller.staggerMutex.Lock()
  2470. timer := time.NewTimer(prng.JitterDuration(staggerPeriod, staggerJitter))
  2471. select {
  2472. case <-timer.C:
  2473. case <-controller.establishCtx.Done():
  2474. }
  2475. timer.Stop()
  2476. controller.staggerMutex.Unlock()
  2477. }
  2478. // Apply any dial rate limit delay now, after unlocking
  2479. // concurrentEstablishTunnelsMutex. The delay may be reduced by the
  2480. // time spent waiting to stagger.
  2481. dialRateLimitDelay -= time.Since(startStagger)
  2482. if dialRateLimitDelay > 0 {
  2483. common.SleepWithContext(controller.establishCtx, dialRateLimitDelay)
  2484. }
  2485. // ConnectTunnel will allocate significant memory, so first attempt to
  2486. // reclaim as much as possible.
  2487. DoGarbageCollection()
  2488. tunnel, err := ConnectTunnel(
  2489. controller.establishCtx,
  2490. controller.config,
  2491. candidateServerEntry.adjustedEstablishStartTime,
  2492. dialParams)
  2493. controller.concurrentEstablishTunnelsMutex.Lock()
  2494. if isIntensive {
  2495. controller.concurrentIntensiveEstablishTunnels -= 1
  2496. }
  2497. controller.concurrentEstablishTunnels -= 1
  2498. controller.concurrentEstablishTunnelsMutex.Unlock()
  2499. // Periodically emit memory metrics during the establishment cycle.
  2500. if !controller.isStopEstablishing() {
  2501. emitMemoryMetrics()
  2502. }
  2503. // Immediately reclaim memory allocated by the establishment. In the case
  2504. // of failure, first clear the reference to the tunnel. In the case of
  2505. // success, the garbage collection may still be effective as the initial
  2506. // phases of some protocols involve significant memory allocation that
  2507. // could now be reclaimed.
  2508. if err != nil {
  2509. tunnel = nil
  2510. }
  2511. DoGarbageCollection()
  2512. if err != nil {
  2513. // Unblock other candidates immediately when server affinity
  2514. // candidate fails.
  2515. if candidateServerEntry.isServerAffinityCandidate {
  2516. close(controller.serverAffinityDoneBroadcast)
  2517. }
  2518. // Before emitting error, check if establish interrupted, in which
  2519. // case the error is noise.
  2520. if controller.isStopEstablishing() {
  2521. break loop
  2522. }
  2523. NoticeInfo("failed to connect to %s: %v",
  2524. candidateServerEntry.serverEntry.GetDiagnosticID(),
  2525. errors.Trace(err))
  2526. continue
  2527. }
  2528. // Deliver connected tunnel.
  2529. // Don't block. Assumes the receiver has a buffer large enough for
  2530. // the number of desired tunnels. If there's no room, the tunnel must
  2531. // not be required so it's discarded.
  2532. select {
  2533. case controller.connectedTunnels <- tunnel:
  2534. default:
  2535. controller.discardTunnel(tunnel)
  2536. // Clear the reference to this discarded tunnel and immediately run
  2537. // a garbage collection to reclaim its memory.
  2538. //
  2539. // Note: this assignment is flagged by github.com/gordonklaus/ineffassign,
  2540. // but should still have some effect on garbage collection?
  2541. tunnel = nil
  2542. DoGarbageCollection()
  2543. }
  2544. // Unblock other candidates only after delivering when
  2545. // server affinity candidate succeeds.
  2546. if candidateServerEntry.isServerAffinityCandidate {
  2547. close(controller.serverAffinityDoneBroadcast)
  2548. }
  2549. }
  2550. }
  2551. func (controller *Controller) isStopEstablishing() bool {
  2552. select {
  2553. case <-controller.establishCtx.Done():
  2554. return true
  2555. default:
  2556. }
  2557. return false
  2558. }
  2559. func (controller *Controller) runInproxyProxy() {
  2560. defer controller.runWaitGroup.Done()
  2561. // Obtain and apply tactics before connecting to the broker and
  2562. // announcing proxies.
  2563. if !controller.config.DisableTunnels &&
  2564. !controller.config.InproxySkipAwaitFullyConnected {
  2565. // When running client tunnel establishment, awaiting establishment
  2566. // guarantees fresh tactics from either an OOB request or a handshake
  2567. // response.
  2568. //
  2569. // While it may be possible to proceed sooner, using cached tactics,
  2570. // waiting until establishment is complete avoids potential races
  2571. // between tactics updates.
  2572. if !controller.awaitFullyEstablished() {
  2573. // Controller is shutting down
  2574. return
  2575. }
  2576. } else {
  2577. // Await the necessary proxy broker specs. These may already be
  2578. // available in cached tactics.
  2579. //
  2580. // When not already available, and when not also running client tunnel
  2581. // establishment, i.e., when DisableTunnels is set,
  2582. // inproxyAwaitProxyBrokerSpecs will perform tactics fetches, in
  2583. // addition to triggering remote server list fetches in case
  2584. // tactics-capable server entries are not available. In this mode,
  2585. // inproxyAwaitProxyBrokerSpecs can return, after a fresh tactics
  2586. // fetch yielding no broker specs, without broker specs.
  2587. // haveInproxyProxyBrokerSpecs is checked again below.
  2588. //
  2589. // InproxySkipAwaitFullyConnected is a special testing case to support
  2590. // server/server_test, where a client must be its own proxy; in this
  2591. // case, awaitFullyEstablished will block forever and can't be used.
  2592. // When InproxySkipAwaitFullyConnected is set and when also running
  2593. // client tunnel establishment, inproxyAwaitProxyBrokerSpecs simply
  2594. // waits until any broker specs become available, which is sufficient
  2595. // for the test but is not as robust as awaiting fresh tactics.
  2596. if !controller.inproxyAwaitProxyBrokerSpecs() {
  2597. // Controller is shutting down
  2598. return
  2599. }
  2600. }
  2601. // Don't announce proxies if tactics indicates it won't be allowed. This
  2602. // is also enforced on the broker; this client-side check cuts down on
  2603. // load from well-behaved proxies.
  2604. //
  2605. // This early check is enforced only when there are tactics, as indicated
  2606. // by presence of a tactics tag. In the proxy-only case where broker
  2607. // specs are shipped in the proxy config, inproxyAwaitProxyBrokerSpecs
  2608. // may return before any tactics are fetched, in which case
  2609. // InproxyAllowProxy will always evaluate to the default, false.
  2610. // inproxyHandleProxyTacticsPayload will check InproxyAllowProxy again,
  2611. // after an initial proxy announce returns fresh tactics.
  2612. p := controller.config.GetParameters().Get()
  2613. disallowProxy := !p.Bool(parameters.InproxyAllowProxy) && p.Tag() != ""
  2614. activityNoticePeriod := p.Duration(parameters.InproxyProxyTotalActivityNoticePeriod)
  2615. p.Close()
  2616. // Running an upstream proxy is also an incompatible case.
  2617. useUpstreamProxy := controller.config.UseUpstreamProxy()
  2618. // In both the awaitFullyEstablished and inproxyAwaitProxyBrokerSpecs
  2619. // cases, we may arrive at this point without broker specs, and must
  2620. // recheck.
  2621. haveBrokerSpecs := haveInproxyProxyBrokerSpecs(controller.config)
  2622. if disallowProxy || useUpstreamProxy || !haveBrokerSpecs || !inproxy.Enabled() {
  2623. if disallowProxy {
  2624. NoticeError("inproxy proxy: not allowed")
  2625. }
  2626. if useUpstreamProxy {
  2627. NoticeError("inproxy proxy: not run due to upstream proxy configuration")
  2628. }
  2629. if !haveBrokerSpecs {
  2630. NoticeError("inproxy proxy: no proxy broker specs")
  2631. }
  2632. if !inproxy.Enabled() {
  2633. NoticeError("inproxy proxy: inproxy implementation is not enabled")
  2634. }
  2635. // Signal failure -- and shutdown -- only if running in proxy-only
  2636. // mode. If also running a tunnel, keep running without proxies.
  2637. if controller.config.DisableTunnels {
  2638. NoticeError("inproxy proxy: aborting")
  2639. controller.SignalComponentFailure()
  2640. }
  2641. return
  2642. }
  2643. // The debugLogging flag is passed to both NoticeCommonLogger and to the
  2644. // inproxy package as well; skipping debug logs in the inproxy package,
  2645. // before calling into the notice logger, avoids unnecessary allocations
  2646. // and formatting when debug logging is off.
  2647. debugLogging := controller.config.InproxyEnableWebRTCDebugLogging
  2648. var lastActivityNotice time.Time
  2649. var lastActivityConnectingClients, lastActivityConnectedClients int32
  2650. var lastActivityConnectingClientsTotal, lastActivityConnectedClientsTotal int32
  2651. var activityTotalBytesUp, activityTotalBytesDown int64
  2652. activityUpdater := func(
  2653. connectingClients int32,
  2654. connectedClients int32,
  2655. bytesUp int64,
  2656. bytesDown int64,
  2657. _ time.Duration) {
  2658. // This emit logic mirrors the logic for NoticeBytesTransferred and
  2659. // NoticeTotalBytesTransferred in tunnel.operateTunnel.
  2660. // InproxyProxyActivity frequently emits bytes transferred since the
  2661. // last notice, when not idle; in addition to the current number of
  2662. // connecting and connected clients, whenever that changes. This
  2663. // frequent notice is excluded from diagnostics and is for UI
  2664. // activity display.
  2665. if controller.config.EmitInproxyProxyActivity &&
  2666. (bytesUp > 0 || bytesDown > 0) ||
  2667. connectingClients != lastActivityConnectingClients ||
  2668. connectedClients != lastActivityConnectedClients {
  2669. NoticeInproxyProxyActivity(
  2670. connectingClients, connectedClients, bytesUp, bytesDown)
  2671. lastActivityConnectingClients = connectingClients
  2672. lastActivityConnectedClients = connectedClients
  2673. }
  2674. activityTotalBytesUp += bytesUp
  2675. activityTotalBytesDown += bytesDown
  2676. // InproxyProxyTotalActivity periodically emits total bytes
  2677. // transferred since starting; in addition to the current number of
  2678. // connecting and connected clients, whenever that changes. This
  2679. // notice is for diagnostics.
  2680. if lastActivityNotice.Add(activityNoticePeriod).Before(time.Now()) ||
  2681. connectingClients != lastActivityConnectingClientsTotal ||
  2682. connectedClients != lastActivityConnectedClientsTotal {
  2683. NoticeInproxyProxyTotalActivity(
  2684. connectingClients, connectedClients,
  2685. activityTotalBytesUp, activityTotalBytesDown)
  2686. lastActivityNotice = time.Now()
  2687. lastActivityConnectingClientsTotal = connectingClients
  2688. lastActivityConnectedClientsTotal = connectedClients
  2689. }
  2690. }
  2691. config := &inproxy.ProxyConfig{
  2692. Logger: NoticeCommonLogger(debugLogging),
  2693. EnableWebRTCDebugLogging: debugLogging,
  2694. WaitForNetworkConnectivity: controller.inproxyWaitForNetworkConnectivity,
  2695. GetCurrentNetworkContext: controller.getCurrentNetworkContext,
  2696. GetBrokerClient: controller.inproxyGetProxyBrokerClient,
  2697. GetBaseAPIParameters: controller.inproxyGetProxyAPIParameters,
  2698. MakeWebRTCDialCoordinator: controller.inproxyMakeProxyWebRTCDialCoordinator,
  2699. HandleTacticsPayload: controller.inproxyHandleProxyTacticsPayload,
  2700. MaxClients: controller.config.InproxyMaxClients,
  2701. LimitUpstreamBytesPerSecond: controller.config.InproxyLimitUpstreamBytesPerSecond,
  2702. LimitDownstreamBytesPerSecond: controller.config.InproxyLimitDownstreamBytesPerSecond,
  2703. MustUpgrade: controller.config.OnInproxyMustUpgrade,
  2704. ActivityUpdater: activityUpdater,
  2705. }
  2706. proxy, err := inproxy.NewProxy(config)
  2707. if err != nil {
  2708. NoticeError("inproxy.NewProxy failed: %v", errors.Trace(err))
  2709. controller.SignalComponentFailure()
  2710. return
  2711. }
  2712. NoticeInfo("inproxy proxy: running")
  2713. proxy.Run(controller.runCtx)
  2714. // Emit one last NoticeInproxyProxyTotalActivity with the final byte counts.
  2715. NoticeInproxyProxyTotalActivity(
  2716. lastActivityConnectingClients, lastActivityConnectedClients,
  2717. activityTotalBytesUp, activityTotalBytesDown)
  2718. NoticeInfo("inproxy proxy: stopped")
  2719. }
  2720. // inproxyAwaitProxyBrokerSpecs awaits proxy broker specs or a fresh tactics
  2721. // fetch indicating that there are no proxy broker specs. The caller should
  2722. // check haveInproxyProxyBrokerSpecs to determine which is the case.
  2723. //
  2724. // inproxyAwaitProxyBrokerSpecs is intended for use either when DisableTunnels
  2725. // is set or when InproxySkipAwaitFullyConnected is set.
  2726. //
  2727. // In the DisableTunnels case, inproxyAwaitProxyBrokerSpecs will perform
  2728. // tactics fetches and trigger remote server list fetches in case
  2729. // tactics-capable server entries are required. The DisableTunnels case
  2730. // assumes client tunnel establishment is not also running, as the tactics
  2731. // operations could otherwise conflict.
  2732. //
  2733. // In the InproxySkipAwaitFullyConnected case, which is intended only to
  2734. // support testing, inproxyAwaitProxyBrokerSpecs simply polls forever for
  2735. // proxy broker specs expected, in the test, to be obtained from concurrent
  2736. // client tunnel establishment operations.
  2737. //
  2738. // inproxyAwaitProxyBrokerSpecs returns false when the Controller is
  2739. // stopping.
  2740. func (controller *Controller) inproxyAwaitProxyBrokerSpecs() bool {
  2741. // Check for any broker specs in cached tactics or config parameters
  2742. // already loaded by NewController or Config.Commit.
  2743. if haveInproxyProxyBrokerSpecs(controller.config) {
  2744. return true
  2745. }
  2746. // If there are no broker specs in config parameters and tactics are
  2747. // disabled, there is nothing more to await.
  2748. if controller.config.DisableTactics {
  2749. NoticeWarning("inproxy proxy: no broker specs and tactics disabled")
  2750. return true
  2751. }
  2752. NoticeInfo("inproxy proxy: await tactics with proxy broker specs")
  2753. // Orchestrating fetches roughly follows the same pattern as
  2754. // establishCandidateGenerator, with a WaitForNetworkConnectivity check,
  2755. // followed by the fetch operation; and a remote server list trigger when
  2756. // that fails, followed by a short pause.
  2757. doFetches := controller.config.DisableTunnels
  2758. // pollPeriod for InproxySkipAwaitFullyConnected case.
  2759. pollPeriod := 100 * time.Millisecond
  2760. for {
  2761. fetched := false
  2762. if doFetches {
  2763. if !WaitForNetworkConnectivity(
  2764. controller.runCtx,
  2765. controller.config.NetworkConnectivityChecker,
  2766. nil) {
  2767. // Controller is shutting down
  2768. return false
  2769. }
  2770. // Force a fetch for the latest tactics, since cached tactics, if
  2771. // any, did not yield proxy broker specs.
  2772. useStoredTactics := false
  2773. fetched = GetTactics(controller.runCtx, controller.config, useStoredTactics)
  2774. }
  2775. if haveInproxyProxyBrokerSpecs(controller.config) {
  2776. return true
  2777. } else if fetched {
  2778. // If fresh tactics yielded no proxy broker specs, there is
  2779. // nothing more to await.
  2780. NoticeWarning("inproxy proxy: no broker specs in tactics")
  2781. return true
  2782. }
  2783. timeout := pollPeriod
  2784. if doFetches {
  2785. // Trigger remote server list fetches in case the tactics fetch
  2786. // failed due to "no capable servers". Repeated triggers will
  2787. // have no effect, subject to FetchRemoteServerListStalePeriod.
  2788. //
  2789. // While triggerFetches also triggers upgrade downloads, currently
  2790. // the upgrade downloader is not enabled when DisableTunnels is
  2791. // set. See Controller.Run.
  2792. //
  2793. // TODO: make the trigger conditional on the specific "no capable
  2794. // servers" failure condition.
  2795. controller.triggerFetches()
  2796. // Pause before attempting to fetch tactics again. This helps
  2797. // avoid some busy wait loop conditions, allows some time for
  2798. // network conditions to change, and also allows for remote server
  2799. // list fetches to complete. The EstablishTunnelPausePeriod and
  2800. // Jitter parameters used in establishCandidateGenerator are also
  2801. // appropriate in this instance.
  2802. p := controller.config.GetParameters().Get()
  2803. timeout = prng.JitterDuration(
  2804. p.Duration(parameters.EstablishTunnelPausePeriod),
  2805. p.Float(parameters.EstablishTunnelPausePeriodJitter))
  2806. p.Close()
  2807. }
  2808. timer := time.NewTimer(timeout)
  2809. select {
  2810. case <-timer.C:
  2811. case <-controller.runCtx.Done():
  2812. timer.Stop()
  2813. // Controller is shutting down
  2814. return false
  2815. }
  2816. timer.Stop()
  2817. }
  2818. }
  2819. func (controller *Controller) inproxyWaitForNetworkConnectivity() bool {
  2820. var isCompatibleNetwork func() bool
  2821. emittedIncompatibleNetworkNotice := false
  2822. if !controller.config.IsInproxyProxyPersonalPairingMode() {
  2823. // Pause announcing proxies when currently running on an incompatible
  2824. // network, such as a non-Psiphon VPN.
  2825. p := controller.config.GetParameters().Get()
  2826. incompatibleNetworkTypes := p.Strings(parameters.InproxyProxyIncompatibleNetworkTypes)
  2827. p.Close()
  2828. isCompatibleNetwork = func() bool {
  2829. compatibleNetwork := !common.Contains(
  2830. incompatibleNetworkTypes,
  2831. GetNetworkType(controller.config.GetNetworkID()))
  2832. if !compatibleNetwork && !emittedIncompatibleNetworkNotice {
  2833. NoticeInfo("inproxy proxy: waiting due to incompatible network")
  2834. emittedIncompatibleNetworkNotice = true
  2835. }
  2836. return compatibleNetwork
  2837. }
  2838. }
  2839. return WaitForNetworkConnectivity(
  2840. controller.runCtx,
  2841. controller.config.NetworkConnectivityChecker,
  2842. isCompatibleNetwork)
  2843. }
  2844. // inproxyGetProxyBrokerClient returns the broker client shared by all proxy
  2845. // operations.
  2846. func (controller *Controller) inproxyGetProxyBrokerClient() (*inproxy.BrokerClient, error) {
  2847. brokerClient, _, err := controller.inproxyProxyBrokerClientManager.GetBrokerClient(
  2848. controller.config.GetNetworkID())
  2849. if err != nil {
  2850. return nil, errors.Trace(err)
  2851. }
  2852. return brokerClient, nil
  2853. }
  2854. func (controller *Controller) inproxyGetProxyAPIParameters(includeTacticsParameters bool) (
  2855. common.APIParameters, string, error) {
  2856. // TODO: include broker fronting dial parameters to be logged by the
  2857. // broker.
  2858. includeSessionID := true
  2859. params := getBaseAPIParameters(
  2860. baseParametersNoDialParameters, nil, includeSessionID, controller.config, nil)
  2861. if controller.config.DisableTactics {
  2862. return params, "", nil
  2863. }
  2864. // Add the stored tactics tag, so that the broker can return new tactics if
  2865. // available.
  2866. //
  2867. // The active network ID is recorded returned and rechecked for
  2868. // consistency when storing any new tactics returned from the broker;
  2869. // other tactics fetches have this same check.
  2870. networkID := controller.config.GetNetworkID()
  2871. if includeTacticsParameters {
  2872. err := tactics.SetTacticsAPIParameters(
  2873. GetTacticsStorer(controller.config), networkID, params)
  2874. if err != nil {
  2875. return nil, "", errors.Trace(err)
  2876. }
  2877. }
  2878. return params, networkID, nil
  2879. }
  2880. func (controller *Controller) inproxyMakeProxyWebRTCDialCoordinator() (
  2881. inproxy.WebRTCDialCoordinator, error) {
  2882. // nil is passed in for both InproxySTUNDialParameters and
  2883. // InproxyWebRTCDialParameters, so those parameters will be newly
  2884. // auto-generated for each client/proxy connection attempt. Unlike the
  2885. // in-proxy client, there is currently no replay of STUN or WebRTC dial
  2886. // parameters.
  2887. isProxy := true
  2888. webRTCDialInstance, err := NewInproxyWebRTCDialInstance(
  2889. controller.config,
  2890. controller.config.GetNetworkID(),
  2891. isProxy,
  2892. controller.inproxyNATStateManager,
  2893. nil,
  2894. nil)
  2895. if err != nil {
  2896. return nil, errors.Trace(err)
  2897. }
  2898. return webRTCDialInstance, nil
  2899. }
  2900. // inproxyHandleProxyTacticsPayload handles new tactics returned from the
  2901. // proxy and returns when tactics have changed.
  2902. //
  2903. // inproxyHandleTacticsPayload duplicates some tactics-handling code from
  2904. // doHandshakeRequest.
  2905. func (controller *Controller) inproxyHandleProxyTacticsPayload(
  2906. networkID string, tacticsPayload []byte) bool {
  2907. if controller.config.DisableTactics {
  2908. return false
  2909. }
  2910. if controller.config.GetNetworkID() != networkID {
  2911. // Ignore the tactics if the network ID has changed.
  2912. return false
  2913. }
  2914. var payload *tactics.Payload
  2915. err := json.Unmarshal(tacticsPayload, &payload)
  2916. if err != nil {
  2917. NoticeError("unmarshal tactics payload failed: %v", errors.Trace(err))
  2918. return false
  2919. }
  2920. if payload == nil {
  2921. // See "null" comment in doHandshakeRequest.
  2922. return false
  2923. }
  2924. // The in-proxy proxy implementation arranges for the first ProxyAnnounce
  2925. // request to get a head start in case there are new tactics available
  2926. // from the broker. Additional requests are also staggered.
  2927. //
  2928. // It can still happen that concurrent in-flight ProxyAnnounce requests
  2929. // receive duplicate new-tactics responses.
  2930. //
  2931. // TODO: detect this case and avoid resetting the broker client and NAT
  2932. // state managers more than necessary.
  2933. // Serialize processing of tactics from ProxyAnnounce responses.
  2934. controller.inproxyHandleTacticsMutex.Lock()
  2935. defer controller.inproxyHandleTacticsMutex.Unlock()
  2936. // When tactics are unchanged, the broker, as in the handshake case,
  2937. // returns a tactics payload, but without new tactics. As in the
  2938. // handshake case, HandleTacticsPayload is called in order to extend the
  2939. // TTL of the locally cached, unchanged tactics. Due to the potential
  2940. // high frequency and concurrency of ProxyAnnnounce requests vs.
  2941. // handshakes, a limit is added to update the data store's tactics TTL no
  2942. // more than one per minute.
  2943. appliedNewTactics := payload.Tactics != nil
  2944. now := time.Now()
  2945. if !appliedNewTactics && now.Sub(controller.inproxyLastStoredTactics) > 1*time.Minute {
  2946. // Skip TTL-only disk write.
  2947. return false
  2948. }
  2949. controller.inproxyLastStoredTactics = now
  2950. tacticsRecord, err := tactics.HandleTacticsPayload(
  2951. GetTacticsStorer(controller.config), networkID, payload)
  2952. if err != nil {
  2953. NoticeError("HandleTacticsPayload failed: %v", errors.Trace(err))
  2954. return false
  2955. }
  2956. if tacticsRecord != nil {
  2957. // SetParameters signals registered components, including broker
  2958. // client and NAT state managers, that must reset upon tactics changes.
  2959. err := controller.config.SetParameters(
  2960. tacticsRecord.Tag, true, tacticsRecord.Tactics.Parameters)
  2961. if err != nil {
  2962. NoticeInfo("apply inproxy broker tactics failed: %s", err)
  2963. return false
  2964. }
  2965. } else {
  2966. appliedNewTactics = false
  2967. }
  2968. if appliedNewTactics {
  2969. // Shutdown if running in proxy-only and tactics now indicate the
  2970. // proxy is not allowed.
  2971. //
  2972. // Limitation: does not immediately stop proxy in dual proxy/tunnel mode.
  2973. p := controller.config.GetParameters().Get()
  2974. disallowProxy := !p.Bool(parameters.InproxyAllowProxy)
  2975. p.Close()
  2976. if disallowProxy {
  2977. NoticeError("inproxy proxy: not allowed")
  2978. if controller.config.DisableTunnels {
  2979. NoticeError("inproxy proxy: shutdown")
  2980. controller.SignalComponentFailure()
  2981. }
  2982. }
  2983. }
  2984. return appliedNewTactics
  2985. }