serverEntry.go 40 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231
  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 protocol
  20. import (
  21. "bufio"
  22. "bytes"
  23. "crypto/ed25519"
  24. "crypto/hmac"
  25. "crypto/rand"
  26. "crypto/sha256"
  27. "encoding/base64"
  28. "encoding/hex"
  29. "encoding/json"
  30. "fmt"
  31. "io"
  32. "net"
  33. "regexp"
  34. "strconv"
  35. "strings"
  36. "sync/atomic"
  37. "time"
  38. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common"
  39. "github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/errors"
  40. )
  41. // ServerEntry represents a Psiphon server. It contains information
  42. // about how to establish a tunnel connection to the server through
  43. // several protocols. Server entries are JSON records downloaded from
  44. // various sources.
  45. type ServerEntry struct {
  46. Tag string `json:"tag,omitempty"`
  47. IpAddress string `json:"ipAddress,omitempty"`
  48. WebServerPort string `json:"webServerPort,omitempty"` // not an int
  49. WebServerSecret string `json:"webServerSecret,omitempty"`
  50. WebServerCertificate string `json:"webServerCertificate,omitempty"`
  51. SshPort int `json:"sshPort,omitempty"`
  52. SshUsername string `json:"sshUsername,omitempty"`
  53. SshPassword string `json:"sshPassword,omitempty"`
  54. SshHostKey string `json:"sshHostKey,omitempty"`
  55. SshObfuscatedPort int `json:"sshObfuscatedPort,omitempty"`
  56. SshObfuscatedQUICPort int `json:"sshObfuscatedQUICPort,omitempty"`
  57. LimitQUICVersions []string `json:"limitQUICVersions,omitempty"`
  58. SshObfuscatedTapDancePort int `json:"sshObfuscatedTapdancePort,omitempty"`
  59. SshObfuscatedConjurePort int `json:"sshObfuscatedConjurePort,omitempty"`
  60. SshObfuscatedKey string `json:"sshObfuscatedKey,omitempty"`
  61. Capabilities []string `json:"capabilities,omitempty"`
  62. Region string `json:"region,omitempty"`
  63. ProviderID string `json:"providerID,omitempty"`
  64. FrontingProviderID string `json:"frontingProviderID,omitempty"`
  65. TlsOSSHPort int `json:"tlsOSSHPort,omitempty"`
  66. MeekServerPort int `json:"meekServerPort,omitempty"`
  67. MeekCookieEncryptionPublicKey string `json:"meekCookieEncryptionPublicKey,omitempty"`
  68. MeekObfuscatedKey string `json:"meekObfuscatedKey,omitempty"`
  69. MeekFrontingHost string `json:"meekFrontingHost,omitempty"`
  70. MeekFrontingHosts []string `json:"meekFrontingHosts,omitempty"`
  71. MeekFrontingDomain string `json:"meekFrontingDomain,omitempty"`
  72. MeekFrontingAddresses []string `json:"meekFrontingAddresses,omitempty"`
  73. MeekFrontingAddressesRegex string `json:"meekFrontingAddressesRegex,omitempty"`
  74. MeekFrontingDisableSNI bool `json:"meekFrontingDisableSNI,omitempty"`
  75. TacticsRequestPublicKey string `json:"tacticsRequestPublicKey,omitempty"`
  76. TacticsRequestObfuscatedKey string `json:"tacticsRequestObfuscatedKey,omitempty"`
  77. ConfigurationVersion int `json:"configurationVersion,omitempty"`
  78. Signature string `json:"signature,omitempty"`
  79. DisableHTTPTransforms bool `json:"disableHTTPTransforms,omitempty"`
  80. DisableObfuscatedQUICTransforms bool `json:"disableObfuscatedQUICTransforms,omitempty"`
  81. DisableOSSHTransforms bool `json:"disableOSSHTransforms,omitempty"`
  82. DisableOSSHPrefix bool `json:"disableOSSHPrefix,omitempty"`
  83. InproxySessionPublicKey string `json:"inproxySessionPublicKey,omitempty"`
  84. InproxySessionRootObfuscationSecret string `json:"inproxySessionRootObfuscationSecret,omitempty"`
  85. InproxySSHPort int `json:"inproxySSHPort,omitempty"`
  86. InproxyOSSHPort int `json:"inproxyOSSHPort,omitempty"`
  87. InproxyQUICPort int `json:"inproxyQUICPort,omitempty"`
  88. InproxyMeekPort int `json:"inproxyMeekPort,omitempty"`
  89. InproxyTlsOSSHPort int `json:"inproxyTlsOSSHPort,omitempty"`
  90. // These local fields are not expected to be present in downloaded server
  91. // entries. They are added by the client to record and report stats about
  92. // how and when server entries are obtained.
  93. // All local fields should be included the list of fields in RemoveUnsignedFields.
  94. LocalSource string `json:"localSource,omitempty"`
  95. LocalTimestamp string `json:"localTimestamp,omitempty"`
  96. IsLocalDerivedTag bool `json:"isLocalDerivedTag,omitempty"`
  97. }
  98. // ServerEntryFields is an alternate representation of ServerEntry which
  99. // enables future compatibility when unmarshaling and persisting new server
  100. // entries which may contain new, unrecognized fields not in the ServerEntry
  101. // type for a particular client version.
  102. //
  103. // When new JSON server entries with new fields are unmarshaled to ServerEntry
  104. // types, unrecognized fields are discarded. When unmarshaled to
  105. // ServerEntryFields, unrecognized fields are retained and may be persisted
  106. // and available when the client is upgraded and unmarshals to an updated
  107. // ServerEntry type.
  108. type ServerEntryFields map[string]interface{}
  109. // GetServerEntry converts a ServerEntryFields into a ServerEntry.
  110. func (fields ServerEntryFields) GetServerEntry() (*ServerEntry, error) {
  111. marshaledServerEntry, err := json.Marshal(fields)
  112. if err != nil {
  113. return nil, errors.Trace(err)
  114. }
  115. var serverEntry *ServerEntry
  116. err = json.Unmarshal(marshaledServerEntry, &serverEntry)
  117. if err != nil {
  118. return nil, errors.Trace(err)
  119. }
  120. return serverEntry, nil
  121. }
  122. func (fields ServerEntryFields) GetTag() string {
  123. tag, ok := fields["tag"]
  124. if !ok {
  125. return ""
  126. }
  127. tagStr, ok := tag.(string)
  128. if !ok {
  129. return ""
  130. }
  131. return tagStr
  132. }
  133. // SetTag sets a local, derived server entry tag. A tag is an identifier used
  134. // in server entry pruning and potentially other use cases. An explict tag,
  135. // set by the Psiphon Network, may be present in a server entry that is
  136. // imported; otherwise, the client will set a derived tag. The tag should be
  137. // generated using GenerateServerEntryTag. When SetTag finds a explicit tag,
  138. // the new, derived tag is ignored. The isLocalTag local field is set to
  139. // distinguish explict and derived tags and is used in signature verification
  140. // to determine if the tag field is part of the signature.
  141. func (fields ServerEntryFields) SetTag(tag string) {
  142. // Don't replace explicit tag
  143. if tag, ok := fields["tag"]; ok {
  144. tagStr, ok := tag.(string)
  145. if ok && tagStr != "" {
  146. isLocalDerivedTag, ok := fields["isLocalDerivedTag"]
  147. if !ok {
  148. return
  149. }
  150. isLocalDerivedTagBool, ok := isLocalDerivedTag.(bool)
  151. if ok && !isLocalDerivedTagBool {
  152. return
  153. }
  154. }
  155. }
  156. fields["tag"] = tag
  157. // Mark this tag as local
  158. fields["isLocalDerivedTag"] = true
  159. }
  160. func (fields ServerEntryFields) GetDiagnosticID() string {
  161. tag, ok := fields["tag"]
  162. if !ok {
  163. return ""
  164. }
  165. tagStr, ok := tag.(string)
  166. if !ok {
  167. return ""
  168. }
  169. return TagToDiagnosticID(tagStr)
  170. }
  171. func (fields ServerEntryFields) GetIPAddress() string {
  172. ipAddress, ok := fields["ipAddress"]
  173. if !ok {
  174. return ""
  175. }
  176. ipAddressStr, ok := ipAddress.(string)
  177. if !ok {
  178. return ""
  179. }
  180. return ipAddressStr
  181. }
  182. func (fields ServerEntryFields) GetWebServerPort() string {
  183. webServerPort, ok := fields["webServerPort"]
  184. if !ok {
  185. return ""
  186. }
  187. webServerPortStr, ok := webServerPort.(string)
  188. if !ok {
  189. return ""
  190. }
  191. return webServerPortStr
  192. }
  193. func (fields ServerEntryFields) GetWebServerSecret() string {
  194. webServerSecret, ok := fields["webServerSecret"]
  195. if !ok {
  196. return ""
  197. }
  198. webServerSecretStr, ok := webServerSecret.(string)
  199. if !ok {
  200. return ""
  201. }
  202. return webServerSecretStr
  203. }
  204. func (fields ServerEntryFields) GetWebServerCertificate() string {
  205. webServerCertificate, ok := fields["webServerCertificate"]
  206. if !ok {
  207. return ""
  208. }
  209. webServerCertificateStr, ok := webServerCertificate.(string)
  210. if !ok {
  211. return ""
  212. }
  213. return webServerCertificateStr
  214. }
  215. func (fields ServerEntryFields) GetConfigurationVersion() int {
  216. configurationVersion, ok := fields["configurationVersion"]
  217. if !ok {
  218. return 0
  219. }
  220. configurationVersionFloat, ok := configurationVersion.(float64)
  221. if !ok {
  222. return 0
  223. }
  224. return int(configurationVersionFloat)
  225. }
  226. func (fields ServerEntryFields) GetLocalSource() string {
  227. localSource, ok := fields["localSource"]
  228. if !ok {
  229. return ""
  230. }
  231. localSourceStr, ok := localSource.(string)
  232. if !ok {
  233. return ""
  234. }
  235. return localSourceStr
  236. }
  237. func (fields ServerEntryFields) SetLocalSource(source string) {
  238. fields["localSource"] = source
  239. }
  240. func (fields ServerEntryFields) GetLocalTimestamp() string {
  241. localTimestamp, ok := fields["localTimestamp"]
  242. if !ok {
  243. return ""
  244. }
  245. localTimestampStr, ok := localTimestamp.(string)
  246. if !ok {
  247. return ""
  248. }
  249. return localTimestampStr
  250. }
  251. func (fields ServerEntryFields) SetLocalTimestamp(timestamp string) {
  252. fields["localTimestamp"] = timestamp
  253. }
  254. func (fields ServerEntryFields) HasSignature() bool {
  255. signature, ok := fields["signature"]
  256. if !ok {
  257. return false
  258. }
  259. signatureStr, ok := signature.(string)
  260. if !ok {
  261. return false
  262. }
  263. return signatureStr != ""
  264. }
  265. const signaturePublicKeyDigestSize = 8
  266. // AddSignature signs a server entry and attaches a new field containing the
  267. // signature. Any existing "signature" field will be replaced.
  268. //
  269. // The signature incudes a public key ID that is derived from a digest of the
  270. // public key value. This ID is intended for future use when multiple signing
  271. // keys may be deployed.
  272. func (fields ServerEntryFields) AddSignature(publicKey, privateKey string) error {
  273. // Make a copy so that removing unsigned fields will have no side effects
  274. copyFields := make(ServerEntryFields)
  275. for k, v := range fields {
  276. copyFields[k] = v
  277. }
  278. copyFields.RemoveUnsignedFields()
  279. delete(copyFields, "signature")
  280. // Best practise would be to sign the JSON encoded server entry bytes and
  281. // append the signature to those bytes. However, due to backwards
  282. // compatibility requirements, we must retain the outer server entry encoding
  283. // as-is and insert the signature.
  284. //
  285. // Limitation: since the verifyier must remarshal its server entry before
  286. // verifying, the JSON produced there must be a byte-for-byte match to the
  287. // JSON signed here. The precise output of the JSON encoder that is used,
  288. // "encoding/json", with default formatting, as of Go 1.11.5, is therefore
  289. // part of the signature protocol.
  290. //
  291. // TODO: use a standard, canonical encoding, such as JCS:
  292. // https://tools.ietf.org/id/draft-rundgren-json-canonicalization-scheme-05.html
  293. marshaledFields, err := json.Marshal(copyFields)
  294. if err != nil {
  295. return errors.Trace(err)
  296. }
  297. decodedPublicKey, err := base64.StdEncoding.DecodeString(publicKey)
  298. if err != nil {
  299. return errors.Trace(err)
  300. }
  301. publicKeyDigest := sha256.Sum256(decodedPublicKey)
  302. publicKeyID := publicKeyDigest[:signaturePublicKeyDigestSize]
  303. decodedPrivateKey, err := base64.StdEncoding.DecodeString(privateKey)
  304. if err != nil {
  305. return errors.Trace(err)
  306. }
  307. signature := ed25519.Sign(decodedPrivateKey, marshaledFields)
  308. fields["signature"] = base64.StdEncoding.EncodeToString(
  309. append(publicKeyID, signature...))
  310. return nil
  311. }
  312. // VerifySignature verifies the signature set by AddSignature.
  313. //
  314. // VerifySignature must be called before using any server entry that is
  315. // imported from an untrusted source, such as client-to-client exchange.
  316. func (fields ServerEntryFields) VerifySignature(publicKey string) error {
  317. if publicKey == "" {
  318. return errors.TraceNew("missing public key")
  319. }
  320. // Make a copy so that removing unsigned fields will have no side effects
  321. copyFields := make(ServerEntryFields)
  322. for k, v := range fields {
  323. copyFields[k] = v
  324. }
  325. signatureField, ok := copyFields["signature"]
  326. if !ok {
  327. return errors.TraceNew("missing signature field")
  328. }
  329. signatureFieldStr, ok := signatureField.(string)
  330. if !ok {
  331. return errors.TraceNew("invalid signature field")
  332. }
  333. decodedSignatureField, err := base64.StdEncoding.DecodeString(signatureFieldStr)
  334. if err != nil {
  335. return errors.Trace(err)
  336. }
  337. if len(decodedSignatureField) < signaturePublicKeyDigestSize {
  338. return errors.TraceNew("invalid signature field length")
  339. }
  340. publicKeyID := decodedSignatureField[:signaturePublicKeyDigestSize]
  341. signature := decodedSignatureField[signaturePublicKeyDigestSize:]
  342. if len(signature) != ed25519.SignatureSize {
  343. return errors.TraceNew("invalid signature length")
  344. }
  345. decodedPublicKey, err := base64.StdEncoding.DecodeString(publicKey)
  346. if err != nil {
  347. return errors.Trace(err)
  348. }
  349. publicKeyDigest := sha256.Sum256(decodedPublicKey)
  350. expectedPublicKeyID := publicKeyDigest[:signaturePublicKeyDigestSize]
  351. if !bytes.Equal(expectedPublicKeyID, publicKeyID) {
  352. return errors.TraceNew("unexpected public key ID")
  353. }
  354. copyFields.RemoveUnsignedFields()
  355. delete(copyFields, "signature")
  356. marshaledFields, err := json.Marshal(copyFields)
  357. if err != nil {
  358. return errors.Trace(err)
  359. }
  360. if !ed25519.Verify(decodedPublicKey, marshaledFields, signature) {
  361. return errors.TraceNew("invalid signature")
  362. }
  363. return nil
  364. }
  365. // RemoveUnsignedFields prepares a server entry for signing or signature
  366. // verification by removing unsigned fields. The JSON marshalling of the
  367. // remaining fields is the data that is signed.
  368. func (fields ServerEntryFields) RemoveUnsignedFields() {
  369. delete(fields, "localSource")
  370. delete(fields, "localTimestamp")
  371. // Only non-local, explicit tags are part of the signature
  372. isLocalDerivedTag := fields["isLocalDerivedTag"]
  373. isLocalDerivedTagBool, ok := isLocalDerivedTag.(bool)
  374. if ok && isLocalDerivedTagBool {
  375. delete(fields, "tag")
  376. }
  377. delete(fields, "isLocalDerivedTag")
  378. }
  379. // ToSignedFields checks for a signature and calls RemoveUnsignedFields.
  380. func (fields ServerEntryFields) ToSignedFields() error {
  381. if !fields.HasSignature() {
  382. return errors.TraceNew("missing signature field")
  383. }
  384. fields.RemoveUnsignedFields()
  385. return nil
  386. }
  387. // NewServerEntrySignatureKeyPair creates an ed25519 key pair for use in
  388. // server entry signing and verification.
  389. func NewServerEntrySignatureKeyPair() (string, string, error) {
  390. publicKey, privateKey, err := ed25519.GenerateKey(rand.Reader)
  391. if err != nil {
  392. return "", "", errors.Trace(err)
  393. }
  394. return base64.StdEncoding.EncodeToString(publicKey),
  395. base64.StdEncoding.EncodeToString(privateKey),
  396. nil
  397. }
  398. // GetCapability returns the server capability corresponding
  399. // to the tunnel protocol.
  400. func GetCapability(protocol string) string {
  401. // The "-OSSH" suffix drop is for legacy compatibility. Newer protocols,
  402. // including in-proxy protocols, use the full protocol name as the
  403. // capability. This avoids ambiguities such as in the case
  404. // of "INPROXY-WEBRTC-OSSH", where a truncated "INPROXY-WEBRTC" is
  405. // ambiguous.
  406. if TunnelProtocolUsesInproxy(protocol) {
  407. return protocol
  408. }
  409. return strings.TrimSuffix(protocol, "-OSSH")
  410. }
  411. // GetTacticsCapability returns the server tactics capability
  412. // corresponding to the tunnel protocol.
  413. func GetTacticsCapability(protocol string) string {
  414. return GetCapability(protocol) + "-TACTICS"
  415. }
  416. // hasCapability indicates if the server entry has the specified capability.
  417. //
  418. // Any internal "PASSTHROUGH-v2 or "PASSTHROUGH" component in the server
  419. // entry's capabilities is ignored. These PASSTHROUGH components are used to
  420. // mask protocols which are running the passthrough mechanisms from older
  421. // clients which do not implement the passthrough messages. Older clients
  422. // will treat these capabilities as unknown protocols and skip them.
  423. //
  424. // Any "QUICv1" capability is treated as "QUIC". "QUICv1" is used to mask the
  425. // QUIC-OSSH capability from older clients to ensure that older clients do
  426. // not send gQUIC packets to second generation QUICv1-only QUIC-OSSH servers.
  427. // New clients must check SupportsOnlyQUICv1 before selecting a QUIC version;
  428. // for "QUICv1", this ensures that new clients also do not select gQUIC to
  429. // QUICv1-only servers.
  430. //
  431. // In-proxy tunnel protocols omit the "v1" and "PASSTHROUGH" suffixes. For
  432. // in-proxy QUIC, gQUIC is never used; and for in-proxy HTTPS/TLS, clients
  433. // always apply PASSTHROUGH-v2.
  434. func (serverEntry *ServerEntry) hasCapability(requiredCapability string) bool {
  435. for _, capability := range serverEntry.Capabilities {
  436. originalCapability := capability
  437. capability = strings.ReplaceAll(capability, "-PASSTHROUGH-v2", "")
  438. capability = strings.ReplaceAll(capability, "-PASSTHROUGH", "")
  439. quicCapability := GetCapability(TUNNEL_PROTOCOL_QUIC_OBFUSCATED_SSH)
  440. if capability == quicCapability+"v1" {
  441. capability = quicCapability
  442. }
  443. if capability == requiredCapability {
  444. return true
  445. }
  446. // Special case: some capabilities may additionally support TLS-OSSH.
  447. // This does not apply to in-proxy TLS-OSSH.
  448. if requiredCapability == GetCapability(TUNNEL_PROTOCOL_TLS_OBFUSCATED_SSH) &&
  449. capabilitySupportsTLSOSSH(originalCapability) {
  450. return true
  451. }
  452. }
  453. return false
  454. }
  455. // capabilitySupportsTLSOSSH returns true if and only if the given capability
  456. // supports TLS-OSSH in addition to its primary protocol.
  457. func capabilitySupportsTLSOSSH(capability string) bool {
  458. tlsCapabilities := []string{
  459. GetCapability(TUNNEL_PROTOCOL_UNFRONTED_MEEK_HTTPS),
  460. GetCapability(TUNNEL_PROTOCOL_UNFRONTED_MEEK_SESSION_TICKET),
  461. }
  462. for _, tlsCapability := range tlsCapabilities {
  463. // The TLS capability is additionally supported by UNFRONTED-MEEK-HTTPS
  464. // and UNFRONTED-MEEK-SESSION-TICKET capabilities with passthrough.
  465. if capability == tlsCapability+"-PASSTHROUGH-v2" {
  466. return true
  467. }
  468. }
  469. return false
  470. }
  471. // SupportsProtocol returns true if and only if the ServerEntry has
  472. // the necessary capability to support the specified tunnel protocol.
  473. func (serverEntry *ServerEntry) SupportsProtocol(protocol string) bool {
  474. requiredCapability := GetCapability(protocol)
  475. return serverEntry.hasCapability(requiredCapability)
  476. }
  477. // ProtocolUsesLegacyPassthrough indicates whether the ServerEntry supports
  478. // the specified protocol using legacy passthrough messages.
  479. //
  480. // There is no corresponding check for v2 passthrough, as clients send v2
  481. // passthrough messages unconditionally, by default, for passthrough
  482. // protocols.
  483. func (serverEntry *ServerEntry) ProtocolUsesLegacyPassthrough(protocol string) bool {
  484. legacyCapability := GetCapability(protocol) + "-PASSTHROUGH"
  485. for _, capability := range serverEntry.Capabilities {
  486. if capability == legacyCapability {
  487. return true
  488. }
  489. }
  490. return false
  491. }
  492. // SupportsOnlyQUICv1 indicates that the QUIC-OSSH server supports only QUICv1
  493. // and gQUIC versions should not be selected, as they will fail to connect
  494. // while sending atypical traffic to the server.
  495. //
  496. // SupportsOnlyQUICv1 strictly applies to QUIC-OSSH and not the in-proxy
  497. // variant.
  498. func (serverEntry *ServerEntry) SupportsOnlyQUICv1() bool {
  499. quicCapability := GetCapability(TUNNEL_PROTOCOL_QUIC_OBFUSCATED_SSH)
  500. return common.Contains(serverEntry.Capabilities, quicCapability+"v1") &&
  501. !common.Contains(serverEntry.Capabilities, quicCapability)
  502. }
  503. // ConditionallyEnabledComponents defines an interface which can be queried to
  504. // determine which conditionally compiled protocol components are present.
  505. type ConditionallyEnabledComponents interface {
  506. QUICEnabled() bool
  507. RefractionNetworkingEnabled() bool
  508. InproxyEnabled() bool
  509. }
  510. // TunnelProtocolPortLists is a map from tunnel protocol names (or "All") to a
  511. // list of port number ranges.
  512. type TunnelProtocolPortLists map[string]*common.PortList
  513. // GetSupportedProtocols returns a list of tunnel protocols supported by the
  514. // ServerEntry's capabilities and allowed by various constraints.
  515. func (serverEntry *ServerEntry) GetSupportedProtocols(
  516. conditionallyEnabled ConditionallyEnabledComponents,
  517. useUpstreamProxy bool,
  518. limitTunnelProtocols TunnelProtocols,
  519. limitTunnelDialPortNumbers TunnelProtocolPortLists,
  520. limitQUICVersions QUICVersions,
  521. excludeIntensive bool,
  522. excludeInproxy bool) TunnelProtocols {
  523. supportedProtocols := make(TunnelProtocols, 0)
  524. for _, tunnelProtocol := range SupportedTunnelProtocols {
  525. if useUpstreamProxy && !TunnelProtocolSupportsUpstreamProxy(tunnelProtocol) {
  526. continue
  527. }
  528. if common.Contains(DisabledTunnelProtocols, tunnelProtocol) {
  529. continue
  530. }
  531. if len(limitTunnelProtocols) > 0 {
  532. if !common.Contains(limitTunnelProtocols, tunnelProtocol) {
  533. continue
  534. }
  535. } else {
  536. if common.Contains(DefaultDisabledTunnelProtocols, tunnelProtocol) {
  537. continue
  538. }
  539. }
  540. if excludeIntensive && TunnelProtocolIsResourceIntensive(tunnelProtocol) {
  541. continue
  542. }
  543. // While in-proxy protocols are TunnelProtocolIsResourceIntensive,
  544. // there's an additional use case for excluding in-proxy protocols as
  545. // controlled by InproxyTunnelProtocolSelectionProbability.
  546. if excludeInproxy && TunnelProtocolUsesInproxy(tunnelProtocol) {
  547. continue
  548. }
  549. if (TunnelProtocolUsesQUIC(tunnelProtocol) &&
  550. !conditionallyEnabled.QUICEnabled()) ||
  551. (TunnelProtocolUsesRefractionNetworking(tunnelProtocol) &&
  552. !conditionallyEnabled.RefractionNetworkingEnabled()) ||
  553. (TunnelProtocolUsesInproxy(tunnelProtocol) &&
  554. !conditionallyEnabled.InproxyEnabled()) {
  555. continue
  556. }
  557. if !serverEntry.SupportsProtocol(tunnelProtocol) {
  558. continue
  559. }
  560. // If the server is limiting QUIC versions, at least one must be
  561. // supported. And if tactics is also limiting QUIC versions, there
  562. // must be a common version in both limit lists for this server entry
  563. // to support QUIC-OSSH.
  564. //
  565. // Limitation: to avoid additional complexity, we do not consider
  566. // DisableFrontingProviderQUICVersion here, as fronting providers are
  567. // expected to support QUICv1 and gQUIC is expected to become
  568. // obsolete in general.
  569. if TunnelProtocolUsesQUIC(tunnelProtocol) && len(serverEntry.LimitQUICVersions) > 0 {
  570. if !common.ContainsAny(serverEntry.LimitQUICVersions, SupportedQUICVersions) {
  571. continue
  572. }
  573. if len(limitQUICVersions) > 0 &&
  574. !common.ContainsAny(serverEntry.LimitQUICVersions, limitQUICVersions) {
  575. continue
  576. }
  577. }
  578. dialPortNumber, err := serverEntry.GetDialPortNumber(tunnelProtocol)
  579. if err != nil {
  580. continue
  581. }
  582. if len(limitTunnelDialPortNumbers) > 0 {
  583. if portList, ok := limitTunnelDialPortNumbers[tunnelProtocol]; ok {
  584. if !portList.Lookup(dialPortNumber) {
  585. continue
  586. }
  587. } else if portList, ok := limitTunnelDialPortNumbers[TUNNEL_PROTOCOLS_ALL]; ok {
  588. if !portList.Lookup(dialPortNumber) {
  589. continue
  590. }
  591. }
  592. }
  593. supportedProtocols = append(supportedProtocols, tunnelProtocol)
  594. }
  595. return supportedProtocols
  596. }
  597. var frontedMeekHTTPSDialPortNumber = int32(443)
  598. // SetFrontedMeekHTTPDialPortNumber sets the FRONTED-MEEK-OSSH dial port
  599. // number, which defaults to 443. Overriding the port number enables running
  600. // test servers where binding to port 443 is not possible.
  601. func SetFrontedMeekHTTPDialPortNumber(port int) {
  602. atomic.StoreInt32(&frontedMeekHTTPSDialPortNumber, int32(port))
  603. }
  604. func (serverEntry *ServerEntry) GetDialPortNumber(tunnelProtocol string) (int, error) {
  605. if !serverEntry.SupportsProtocol(tunnelProtocol) {
  606. return 0, errors.TraceNew("protocol not supported")
  607. }
  608. if !TunnelProtocolUsesInproxy(tunnelProtocol) {
  609. switch tunnelProtocol {
  610. case TUNNEL_PROTOCOL_TLS_OBFUSCATED_SSH:
  611. if serverEntry.TlsOSSHPort == 0 {
  612. // Special case: a server which supports UNFRONTED-MEEK-HTTPS-OSSH
  613. // or UNFRONTED-MEEK-SESSION-TICKET-OSSH also supports TLS-OSSH
  614. // over the same port.
  615. return serverEntry.MeekServerPort, nil
  616. }
  617. return serverEntry.TlsOSSHPort, nil
  618. case TUNNEL_PROTOCOL_SSH:
  619. return serverEntry.SshPort, nil
  620. case TUNNEL_PROTOCOL_OBFUSCATED_SSH:
  621. return serverEntry.SshObfuscatedPort, nil
  622. case TUNNEL_PROTOCOL_TAPDANCE_OBFUSCATED_SSH:
  623. return serverEntry.SshObfuscatedTapDancePort, nil
  624. case TUNNEL_PROTOCOL_CONJURE_OBFUSCATED_SSH:
  625. return serverEntry.SshObfuscatedConjurePort, nil
  626. case TUNNEL_PROTOCOL_QUIC_OBFUSCATED_SSH:
  627. return serverEntry.SshObfuscatedQUICPort, nil
  628. case TUNNEL_PROTOCOL_FRONTED_MEEK,
  629. TUNNEL_PROTOCOL_FRONTED_MEEK_QUIC_OBFUSCATED_SSH:
  630. return int(atomic.LoadInt32(&frontedMeekHTTPSDialPortNumber)), nil
  631. case TUNNEL_PROTOCOL_FRONTED_MEEK_HTTP:
  632. return 80, nil
  633. case TUNNEL_PROTOCOL_UNFRONTED_MEEK_HTTPS,
  634. TUNNEL_PROTOCOL_UNFRONTED_MEEK_SESSION_TICKET,
  635. TUNNEL_PROTOCOL_UNFRONTED_MEEK:
  636. return serverEntry.MeekServerPort, nil
  637. }
  638. } else {
  639. // Distinct dial/listening ports are used for tunnel protocols when
  640. // used as an in-proxy 2nd hop, as the server will require a relayed
  641. // in-proxy broker report for in-proxy 2nd hops.
  642. switch TunnelProtocolMinusInproxy(tunnelProtocol) {
  643. case TUNNEL_PROTOCOL_TLS_OBFUSCATED_SSH:
  644. if serverEntry.InproxyTlsOSSHPort == 0 {
  645. return serverEntry.InproxyMeekPort, nil
  646. }
  647. return serverEntry.InproxyTlsOSSHPort, nil
  648. case TUNNEL_PROTOCOL_SSH:
  649. return serverEntry.InproxySSHPort, nil
  650. case TUNNEL_PROTOCOL_OBFUSCATED_SSH:
  651. return serverEntry.InproxyOSSHPort, nil
  652. case TUNNEL_PROTOCOL_QUIC_OBFUSCATED_SSH:
  653. return serverEntry.InproxyQUICPort, nil
  654. case TUNNEL_PROTOCOL_FRONTED_MEEK,
  655. TUNNEL_PROTOCOL_FRONTED_MEEK_QUIC_OBFUSCATED_SSH:
  656. return int(atomic.LoadInt32(&frontedMeekHTTPSDialPortNumber)), nil
  657. case TUNNEL_PROTOCOL_FRONTED_MEEK_HTTP:
  658. return 80, nil
  659. case TUNNEL_PROTOCOL_UNFRONTED_MEEK_HTTPS,
  660. TUNNEL_PROTOCOL_UNFRONTED_MEEK_SESSION_TICKET,
  661. TUNNEL_PROTOCOL_UNFRONTED_MEEK:
  662. return serverEntry.InproxyMeekPort, nil
  663. }
  664. }
  665. return 0, errors.TraceNew("unknown protocol")
  666. }
  667. // GetTLSSessionCacheKeyAddress returns a network address (IP:port) that is
  668. // suitable to use as a TLS session cache key.
  669. //
  670. // By default, TLS implementations, including crypto/tls and utls use SNI as a
  671. // session cache key, but this is not a suitable key when SNI is manipulated.
  672. // When SNI is not present, these implementations fall back to using the peer
  673. // remote address, which is also not a suitable key in cases where there is a
  674. // non-TLS-terminating temporary intermediary, such as an in-proxy proxy.
  675. //
  676. // The key is unique to the Psiphon server and tunnel protocol listener. For
  677. // direct tunnel protocols, the key precisely maps TLS sessions to the
  678. // corresponding TLS server. For indirect tunnel protocols, with an
  679. // intermediate TLS server, the key is an approximate map which assumes the
  680. // redials will mostly use the same intermediate TLS server.
  681. //
  682. // Do not use the GetTLSSessionCacheKeyAddress value for dialing.
  683. func (serverEntry *ServerEntry) GetTLSSessionCacheKeyAddress(tunnelProtocol string) (string, error) {
  684. port, err := serverEntry.GetDialPortNumber(tunnelProtocol)
  685. if err != nil {
  686. return "", errors.Trace(err)
  687. }
  688. return net.JoinHostPort(serverEntry.IpAddress, strconv.Itoa(port)), nil
  689. }
  690. // IsValidInproxyDialAddress indicates whether the dial destination
  691. // network/host/port matches the dial parameters for any of the tunnel
  692. // protocols supported by the server entry.
  693. //
  694. // Limitations:
  695. // - TAPDANCE-OSSH and CONJURE-OSSH are not supported.
  696. // - The host header is not considered in the case of fronted protocols.
  697. func (serverEntry *ServerEntry) IsValidInproxyDialAddress(
  698. networkProtocol string, dialHost string, dialPortNumber int) bool {
  699. // The TapDance and Conjure destination addresses are not included
  700. // in the server entry, so TAPDANCE-OSSH and CONJURE-OSSH dial
  701. // destinations cannot be validated for in-proxy use.
  702. for _, tunnelProtocol := range SupportedTunnelProtocols {
  703. if !TunnelProtocolUsesInproxy(tunnelProtocol) {
  704. continue
  705. }
  706. if !serverEntry.SupportsProtocol(tunnelProtocol) {
  707. continue
  708. }
  709. usesTCP := TunnelProtocolUsesTCP(tunnelProtocol)
  710. if (usesTCP && networkProtocol != "tcp") || (!usesTCP && networkProtocol != "udp") {
  711. continue
  712. }
  713. tunnelPortNumber, err := serverEntry.GetDialPortNumber(tunnelProtocol)
  714. if err != nil || tunnelPortNumber != dialPortNumber {
  715. // Silently fail on error as the server entry should be well-formed.
  716. continue
  717. }
  718. if !TunnelProtocolUsesFrontedMeek(tunnelProtocol) {
  719. // For all direct protocols, the destination host must be the
  720. // server IP address.
  721. if serverEntry.IpAddress != dialHost {
  722. continue
  723. }
  724. } else {
  725. // For fronted protocols, the destination host may be domain and
  726. // must match either MeekFrontingAddressesRegex or
  727. // MeekFrontingAddresses. As in psiphon.selectFrontingParameters,
  728. // MeekFrontingAddressesRegex takes precedence when not empty.
  729. //
  730. // As the host header value is not checked here, additional
  731. // measures must be taken to ensure the destination is a Psiphon server.
  732. if len(serverEntry.MeekFrontingAddressesRegex) > 0 {
  733. re, err := regexp.Compile(serverEntry.MeekFrontingAddressesRegex)
  734. if err != nil {
  735. continue
  736. }
  737. // The entire dialHost string must match the regex.
  738. re.Longest()
  739. match := re.FindString(dialHost)
  740. if match == "" || match != dialHost {
  741. continue
  742. }
  743. } else {
  744. if !common.Contains(serverEntry.MeekFrontingAddresses, dialHost) {
  745. continue
  746. }
  747. }
  748. }
  749. // When all of the checks pass for this protocol, the input is a valid
  750. // dial destination.
  751. return true
  752. }
  753. return false
  754. }
  755. // GetSupportedTacticsProtocols returns a list of tunnel protocols,
  756. // supported by the ServerEntry's capabilities, that may be used
  757. // for tactics requests.
  758. func (serverEntry *ServerEntry) GetSupportedTacticsProtocols() []string {
  759. supportedProtocols := make([]string, 0)
  760. for _, protocol := range SupportedTunnelProtocols {
  761. if !TunnelProtocolUsesMeek(protocol) {
  762. continue
  763. }
  764. requiredCapability := GetTacticsCapability(protocol)
  765. if !serverEntry.hasCapability(requiredCapability) {
  766. continue
  767. }
  768. supportedProtocols = append(supportedProtocols, protocol)
  769. }
  770. return supportedProtocols
  771. }
  772. // SupportsSSHAPIRequests returns true when the server supports
  773. // SSH API requests.
  774. func (serverEntry *ServerEntry) SupportsSSHAPIRequests() bool {
  775. return serverEntry.hasCapability(CAPABILITY_SSH_API_REQUESTS)
  776. }
  777. func (serverEntry *ServerEntry) HasProviderID() bool {
  778. return serverEntry.ProviderID != ""
  779. }
  780. func (serverEntry *ServerEntry) HasSignature() bool {
  781. return serverEntry.Signature != ""
  782. }
  783. func (serverEntry *ServerEntry) GetDiagnosticID() string {
  784. return TagToDiagnosticID(serverEntry.Tag)
  785. }
  786. // GenerateServerEntryTag creates a server entry tag value that is
  787. // cryptographically derived from the IP address and web server secret in a
  788. // way that is difficult to reverse the IP address value from the tag or
  789. // compute the tag without having the web server secret, a 256-bit random
  790. // value which is unique per server, in addition to the IP address. A database
  791. // consisting only of server entry tags should be resistent to an attack that
  792. // attempts to reverse all the server IPs, even given a small IP space (IPv4),
  793. // or some subset of the web server secrets.
  794. func GenerateServerEntryTag(ipAddress, webServerSecret string) string {
  795. h := hmac.New(sha256.New, []byte(webServerSecret))
  796. h.Write([]byte(ipAddress))
  797. return base64.StdEncoding.EncodeToString(h.Sum(nil))
  798. }
  799. // TagToDiagnosticID returns a prefix of the server entry tag that should be
  800. // sufficient to uniquely identify servers in diagnostics, while also being
  801. // more human readable than emitting the full tag. The tag is used as the base
  802. // of the diagnostic ID as it doesn't leak the server IP address in diagnostic
  803. // output.
  804. func TagToDiagnosticID(tag string) string {
  805. if len(tag) < 8 {
  806. return "<unknown>"
  807. }
  808. return tag[:8]
  809. }
  810. // EncodeServerEntry returns a string containing the encoding of
  811. // a ServerEntry following Psiphon conventions.
  812. func EncodeServerEntry(serverEntry *ServerEntry) (string, error) {
  813. encodedServerEntry, err := encodeServerEntry(
  814. serverEntry.IpAddress,
  815. serverEntry.WebServerPort,
  816. serverEntry.WebServerSecret,
  817. serverEntry.WebServerCertificate,
  818. serverEntry)
  819. if err != nil {
  820. return "", errors.Trace(err)
  821. }
  822. return encodedServerEntry, nil
  823. }
  824. // EncodeServerEntryFields returns a string containing the encoding of
  825. // ServerEntryFields following Psiphon conventions.
  826. func EncodeServerEntryFields(serverEntryFields ServerEntryFields) (string, error) {
  827. encodedServerEntry, err := encodeServerEntry(
  828. serverEntryFields.GetIPAddress(),
  829. serverEntryFields.GetWebServerPort(),
  830. serverEntryFields.GetWebServerSecret(),
  831. serverEntryFields.GetWebServerCertificate(),
  832. serverEntryFields)
  833. if err != nil {
  834. return "", errors.Trace(err)
  835. }
  836. return encodedServerEntry, nil
  837. }
  838. func encodeServerEntry(
  839. prefixIPAddress string,
  840. prefixWebServerPort string,
  841. prefixWebServerSecret string,
  842. prefixWebServerCertificate string,
  843. serverEntry interface{}) (string, error) {
  844. serverEntryJSON, err := json.Marshal(serverEntry)
  845. if err != nil {
  846. return "", errors.Trace(err)
  847. }
  848. // Legacy clients use a space-delimited fields prefix, and all clients expect
  849. // to at least parse the prefix in order to skip over it.
  850. //
  851. // When the server entry has no web API server certificate, the entire prefix
  852. // can be compacted down to single character placeholders. Clients that can
  853. // use the ssh API always prefer it over the web API and won't use the prefix
  854. // values.
  855. if len(prefixWebServerCertificate) == 0 {
  856. prefixIPAddress = "0"
  857. prefixWebServerPort = "0"
  858. prefixWebServerSecret = "0"
  859. prefixWebServerCertificate = "0"
  860. }
  861. return hex.EncodeToString([]byte(fmt.Sprintf(
  862. "%s %s %s %s %s",
  863. prefixIPAddress,
  864. prefixWebServerPort,
  865. prefixWebServerSecret,
  866. prefixWebServerCertificate,
  867. serverEntryJSON))), nil
  868. }
  869. // DecodeServerEntry extracts a server entry from the encoding
  870. // used by remote server lists and Psiphon server handshake requests.
  871. //
  872. // The resulting ServerEntry.LocalSource is populated with serverEntrySource,
  873. // which should be one of SERVER_ENTRY_SOURCE_EMBEDDED, SERVER_ENTRY_SOURCE_REMOTE,
  874. // SERVER_ENTRY_SOURCE_DISCOVERY, SERVER_ENTRY_SOURCE_TARGET,
  875. // SERVER_ENTRY_SOURCE_OBFUSCATED.
  876. // ServerEntry.LocalTimestamp is populated with the provided timestamp, which
  877. // should be a RFC 3339 formatted string. These local fields are stored with the
  878. // server entry and reported to the server as stats (a coarse granularity timestamp
  879. // is reported).
  880. func DecodeServerEntry(
  881. encodedServerEntry, timestamp, serverEntrySource string) (*ServerEntry, error) {
  882. serverEntry := new(ServerEntry)
  883. err := decodeServerEntry(encodedServerEntry, timestamp, serverEntrySource, serverEntry)
  884. if err != nil {
  885. return nil, errors.Trace(err)
  886. }
  887. // NOTE: if the source JSON happens to have values in these fields, they get clobbered.
  888. serverEntry.LocalSource = serverEntrySource
  889. serverEntry.LocalTimestamp = timestamp
  890. return serverEntry, nil
  891. }
  892. // DecodeServerEntryFields extracts an encoded server entry into a
  893. // ServerEntryFields type, much like DecodeServerEntry. Unrecognized fields
  894. // not in ServerEntry are retained in the ServerEntryFields.
  895. //
  896. // LocalSource/LocalTimestamp map entries are set only when the corresponding
  897. // inputs are non-blank.
  898. func DecodeServerEntryFields(
  899. encodedServerEntry, timestamp, serverEntrySource string) (ServerEntryFields, error) {
  900. serverEntryFields := make(ServerEntryFields)
  901. err := decodeServerEntry(encodedServerEntry, timestamp, serverEntrySource, &serverEntryFields)
  902. if err != nil {
  903. return nil, errors.Trace(err)
  904. }
  905. // NOTE: if the source JSON happens to have values in these fields, they get clobbered.
  906. if serverEntrySource != "" {
  907. serverEntryFields.SetLocalSource(serverEntrySource)
  908. }
  909. if timestamp != "" {
  910. serverEntryFields.SetLocalTimestamp(timestamp)
  911. }
  912. return serverEntryFields, nil
  913. }
  914. func decodeServerEntry(
  915. encodedServerEntry, timestamp, serverEntrySource string,
  916. target interface{}) error {
  917. hexDecodedServerEntry, err := hex.DecodeString(encodedServerEntry)
  918. if err != nil {
  919. return errors.Trace(err)
  920. }
  921. // Skip past legacy format (4 space delimited fields) and just parse the JSON config
  922. fields := bytes.SplitN(hexDecodedServerEntry, []byte(" "), 5)
  923. if len(fields) != 5 {
  924. return errors.TraceNew("invalid encoded server entry")
  925. }
  926. err = json.Unmarshal(fields[4], target)
  927. if err != nil {
  928. return errors.Trace(err)
  929. }
  930. return nil
  931. }
  932. // ValidateServerEntryFields checks for malformed server entries.
  933. func ValidateServerEntryFields(serverEntryFields ServerEntryFields) error {
  934. // Checks for a valid ipAddress. This is important since the IP
  935. // address is the key used to store/lookup the server entry.
  936. ipAddress := serverEntryFields.GetIPAddress()
  937. if net.ParseIP(ipAddress) == nil {
  938. return errors.Tracef("server entry has invalid ipAddress: %s", ipAddress)
  939. }
  940. // TODO: validate more fields?
  941. // Ensure locally initialized fields have been set.
  942. source := serverEntryFields.GetLocalSource()
  943. if !common.Contains(
  944. SupportedServerEntrySources, source) {
  945. return errors.Tracef("server entry has invalid source: %s", source)
  946. }
  947. timestamp := serverEntryFields.GetLocalTimestamp()
  948. _, err := time.Parse(time.RFC3339, timestamp)
  949. if err != nil {
  950. return errors.Tracef("server entry has invalid timestamp: %s", err)
  951. }
  952. return nil
  953. }
  954. // DecodeServerEntryList extracts server entries from the list encoding
  955. // used by remote server lists and Psiphon server handshake requests.
  956. // Each server entry is validated and invalid entries are skipped.
  957. // See DecodeServerEntry for note on serverEntrySource/timestamp.
  958. func DecodeServerEntryList(
  959. encodedServerEntryList, timestamp,
  960. serverEntrySource string) ([]ServerEntryFields, error) {
  961. serverEntries := make([]ServerEntryFields, 0)
  962. for _, encodedServerEntry := range strings.Split(encodedServerEntryList, "\n") {
  963. if len(encodedServerEntry) == 0 {
  964. continue
  965. }
  966. // TODO: skip this entry and continue if can't decode?
  967. serverEntryFields, err := DecodeServerEntryFields(encodedServerEntry, timestamp, serverEntrySource)
  968. if err != nil {
  969. return nil, errors.Trace(err)
  970. }
  971. if ValidateServerEntryFields(serverEntryFields) != nil {
  972. // Skip this entry and continue with the next one
  973. // TODO: invoke a logging callback
  974. continue
  975. }
  976. serverEntries = append(serverEntries, serverEntryFields)
  977. }
  978. return serverEntries, nil
  979. }
  980. // StreamingServerEntryDecoder performs the DecodeServerEntryList
  981. // operation, loading only one server entry into memory at a time.
  982. type StreamingServerEntryDecoder struct {
  983. scanner *bufio.Scanner
  984. timestamp string
  985. serverEntrySource string
  986. }
  987. // NewStreamingServerEntryDecoder creates a new StreamingServerEntryDecoder.
  988. func NewStreamingServerEntryDecoder(
  989. encodedServerEntryListReader io.Reader,
  990. timestamp, serverEntrySource string) *StreamingServerEntryDecoder {
  991. return &StreamingServerEntryDecoder{
  992. scanner: bufio.NewScanner(encodedServerEntryListReader),
  993. timestamp: timestamp,
  994. serverEntrySource: serverEntrySource,
  995. }
  996. }
  997. // Next reads and decodes, and validates the next server entry from the
  998. // input stream, returning a nil server entry when the stream is complete.
  999. //
  1000. // Limitations:
  1001. // - Each encoded server entry line cannot exceed bufio.MaxScanTokenSize,
  1002. // the default buffer size which this decoder uses. This is 64K.
  1003. // - DecodeServerEntry is called on each encoded server entry line, which
  1004. // will allocate memory to hex decode and JSON deserialze the server
  1005. // entry. As this is not presently reusing a fixed buffer, each call
  1006. // will allocate additional memory; garbage collection is necessary to
  1007. // reclaim that memory for reuse for the next server entry.
  1008. func (decoder *StreamingServerEntryDecoder) Next() (ServerEntryFields, error) {
  1009. for {
  1010. if !decoder.scanner.Scan() {
  1011. return nil, errors.Trace(decoder.scanner.Err())
  1012. }
  1013. // TODO: use scanner.Bytes which doesn't allocate, instead of scanner.Text
  1014. // TODO: skip this entry and continue if can't decode?
  1015. serverEntryFields, err := DecodeServerEntryFields(
  1016. decoder.scanner.Text(), decoder.timestamp, decoder.serverEntrySource)
  1017. if err != nil {
  1018. return nil, errors.Trace(err)
  1019. }
  1020. if ValidateServerEntryFields(serverEntryFields) != nil {
  1021. // Skip this entry and continue with the next one
  1022. // TODO: invoke a logging callback
  1023. continue
  1024. }
  1025. return serverEntryFields, nil
  1026. }
  1027. }