Compare commits

..

16 Commits

Author SHA1 Message Date
Sudarsan Reddy 9cd2780079 Release 2022.2.2 2022-02-22 17:31:06 +00:00
Sudarsan Reddy 9909e9d63c TUN-5754: Allow ingress validate to take plaintext option
Ingress validate currently validates config from a file. This PR adds a
new --json/-j flag to provide the ingress/config data as a plaintext
command line argument.
2022-02-22 16:56:22 +00:00
João Oliveirinha 051b2cf352 TUN-5678: Cloudflared uses typed tunnel API 2022-02-21 17:11:31 +00:00
Nuno Diegues 4cf462e322 Release 2022.2.1 2022-02-18 09:19:44 +00:00
cthuang e56c4532ce TUN-5697: Listen for UpdateConfiguration RPC in quic transport 2022-02-18 09:12:39 +00:00
João Oliveirinha d78a5ba5da Revert "TUN-5678: Adapt cloudflared to use new typed APIs"
This reverts commit 630650b90e.
2022-02-17 18:56:07 +00:00
Areg Harutyunyan 1c50618f97 Merge remote-tracking branch 'upstream/master' 2022-02-17 12:16:05 -06:00
João Oliveirinha 630650b90e TUN-5678: Adapt cloudflared to use new typed APIs 2022-02-17 12:15:21 -06:00
cthuang d68ff390ca TUN-5698: Make ingress rules and warp routing dynamically configurable 2022-02-16 09:38:28 +00:00
abelinkinbio 0571210374 TUN-5768: Update cloudflared license file 2022-02-15 10:15:52 +00:00
abe a16dee1d2a Update issue templates 2022-02-14 14:02:45 -06:00
abe 2f6f865f92 Update issue templates 2022-02-14 14:01:17 -06:00
cthuang e22422aafb TUN-5749: Refactor cloudflared to pave way for reconfigurable ingress
- Split origin into supervisor and proxy packages
- Create configManager to handle dynamic config
2022-02-14 15:37:09 +00:00
Nuno Diegues ff4cfeda0c TUN-5184: Handle errors in bidrectional streaming (websocket#Stream) gracefully when 1 side has ended 2022-02-10 20:00:36 +00:00
cthuang db01127191 TUN-5184: Make sure outstanding websocket write is finished, and no more writes after shutdown 2022-02-10 09:43:52 +00:00
cthuang 1ff5fd3fdc TUN-5744: Add a test to make sure cloudflared uses scheme defined in ingress rule, not X-Forwarded-Proto header 2022-02-04 16:51:37 +00:00
50 changed files with 1818 additions and 518 deletions
+31
View File
@@ -0,0 +1,31 @@
---
name: "Bug report \U0001F41B"
about: Create a report to help us improve cloudflared
title: ''
labels: awaiting reply, bug
assignees: ''
---
**Describe the bug**
A clear and concise description of what the bug is.
**To Reproduce**
Steps to reproduce the behavior:
1. Configure '...'
2. Run '....'
3. See error
**Expected behavior**
A clear and concise description of what you expected to happen.
**Environment and versions**
- OS: [e.g. MacOS]
- Architecture: [e.g. AMD, ARM]
- Version: [e.g. 2022.02.0]
**Logs and errors**
If applicable, add logs or errors to help explain your problem.
**Additional context**
Add any other context about the problem here.
@@ -0,0 +1,17 @@
---
name: "Feature request \U0001F4A1"
about: Suggest a feature or enhancement for cloudflared
title: ''
labels: awaiting reply, feature-request
assignees: ''
---
**Describe the feature you'd like**
A clear and concise description of the feature. What problem does it solve for you?
**Describe alternatives you've considered**
Are there any alternatives to solving this problem? If so, what was your experience with them?
**Additional context**
Add any other context or screenshots about the feature request here.
+189 -133
View File
@@ -1,155 +1,211 @@
SERVICES AGREEMENT
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/
Your installation of this software is symbol of your signature indicating that
you accept the terms of this Services Agreement (this "Agreement"). This
Agreement is a legal agreement between you (either an individual or a single
entity) and CloudFlare, Inc. for the services being provided to you by
CloudFlare or its authorized representative (the "Services"), including any
computer software and any associated media, printed materials, and "online" or
electronic documentation provided in connection with the Services (the
"Software" and together with the Services are hereinafter collectively referred
to as the "Solution"). If the user is not an individual, then "you" means your
company, its officers, members, employees, agents, representatives, successors
and assigns. BY USING THE SOLUTION, YOU ARE INDICATING THAT YOU HAVE READ, AND
AGREE TO BE BOUND BY, THE POLICIES, TERMS, AND CONDITIONS SET FORTH BELOW IN
THEIR ENTIRETY WITHOUT LIMITATION OR QUALIFICATION, AS WELL AS BY ALL APPLICABLE
LAWS AND REGULATIONS, AS IF YOU HAD HANDWRITTEN YOUR NAME ON A CONTRACT. IF YOU
DO NOT AGREE TO THESE TERMS AND CONDITIONS, YOU MAY NOT USE THE SOLUTION.
TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
1. GRANT OF RIGHTS
1. Definitions.
1.1 Grant of License. The Solution is licensed by CloudFlare and its
licensors, not sold. Subject to the terms and conditions of this Agreement,
CloudFlare hereby grants you a nonexclusive, nonsublicensable, nontransferable
license to use the Solution. You may examine source code, if provided to you,
solely for the limited purpose of evaluating the Software for security flaws.
You may also use the Service to create derivative works which are exclusively
compatible with any CloudFlare product serviceand no other product or service.
This license applies to the parts of the Solution developed by CloudFlare. The
Solution may also incorporate externally maintained libraries and other open software.
These resources may be governed by other licenses.
"License" shall mean the terms and conditions for use, reproduction,
and distribution as defined by Sections 1 through 9 of this document.
1.2 Restrictions. The license granted herein is granted solely to you and
not, by implication or otherwise, to any of your parents, subsidiaries or
affiliates. No right is granted hereunder to use the Solution to perform
services for third parties. All rights not expressly granted hereunder are
reserved to CloudFlare. You may not use the Solution except as explicitly
permitted under this Agreement. You are expressly prohibited from modifying,
adapting, translating, preparing derivative works from, decompiling, reverse
engineering, disassembling or otherwise attempting to derive source code from
the Software used to provide the Services or any internal data files generated
by the Solution. You are also prohibited from removing, obscuring or altering
any copyright notice, trademarks, or other proprietary rights notices affixed to
or associated with the Solution.
"Licensor" shall mean the copyright owner or entity authorized by
the copyright owner that is granting the License.
1.3 Ownership. As between the parties, CloudFlare and/or its licensors own
and shall retain all right, title, and interest in and to the Solution,
including any and all technology embodied therein, including all copyrights,
patents, trade secrets, trade dress and other proprietary rights associated
therewith, and any derivative works created there from.
"Legal Entity" shall mean the union of the acting entity and all
other entities that control, are controlled by, or are under common
control with that entity. For the purposes of this definition,
"control" means (i) the power, direct or indirect, to cause the
direction or management of such entity, whether by contract or
otherwise, or (ii) ownership of fifty percent (50%) or more of the
outstanding shares, or (iii) beneficial ownership of such entity.
2. LIMITATION OF LIABILITY
"You" (or "Your") shall mean an individual or Legal Entity
exercising permissions granted by this License.
YOU EXPRESSLY ACKNOWLEDGE AND AGREE THAT DOWNLOADING THE SOFTWARE IS AT YOUR
SOLE RISK. THE SOFTWARE IS PROVIDED "AS IS" AND WITHOUT WARRANTY OF ANY KIND
AND CLOUDFLARE, ITS LICENSORS AND ITS AUTHORIZED REPRESENTATIVES (TOGETHER FOR
PURPOSES HEREOF, "CLOUDFLARE") EXPRESSLY DISCLAIM ALL WARRANTIES, EXPRESS OR
IMPLIED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. CLOUDFLARE DOES NOT
WARRANT THAT THE FUNCTIONS CONTAINED IN THE SOFTWARE WILL MEET YOUR
REQUIREMENTS, OR THAT THE OPERATION OF THE SOFTWARE WILL BE UNINTERRUPTED OR
ERROR-FREE, OR THAT DEFECTS IN THE SOFTWARE WILL BE CORRECTED. FURTHERMORE,
CLOUDFLARE DOES NOT WARRANT OR MAKE ANY REPRESENTATIONS REGARDING THE SOFTWARE
OR RELATED DOCUMENTATION IN TERMS OF THEIR CORRECTNESS, ACCURACY, RELIABILITY,
OR OTHERWISE. NO ORAL OR WRITTEN INFORMATION OR ADVICE GIVEN BY CLOUDFLARE SHALL
CREATE A WARRANTY OR IN ANY WAY INCREASE THE SCOPE OF THIS WARRANTY.
"Source" form shall mean the preferred form for making modifications,
including but not limited to software source code, documentation
source, and configuration files.
3. CONFIDENTIALITY
"Object" form shall mean any form resulting from mechanical
transformation or translation of a Source form, including but
not limited to compiled object code, generated documentation,
and conversions to other media types.
It may be necessary during the set up and performance of the Solution for the
parties to exchange Confidential Information. "Confidential Information" means
any information whether oral, or written, of a private, secret, proprietary or
confidential nature, concerning either party or its business operations,
including without limitation: (a) your data and (b) CloudFlare's access control
systems, specialized network equipment and techniques related to the Solution,
use policies, which include trade secrets of CloudFlare and its licensors. Each
party agrees to use the same degree of care to protect the confidentiality of
the Confidential Information of the other party and to prevent its unauthorized
use or dissemination as it uses to protect its own Confidential Information of a
similar nature, but in no event shall exercise less than due diligence and
reasonable care. Each party agrees to use the Confidential Information of the
other party only for purposes related to the performance of this Agreement. All
Confidential Information remains the property of the party disclosing the
information and no license or other rights to Confidential Information is
granted or implied hereby.
"Work" shall mean the work of authorship, whether in Source or
Object form, made available under the License, as indicated by a
copyright notice that is included in or attached to the work
(an example is provided in the Appendix below).
4. TERM AND TERMINATION
"Derivative Works" shall mean any work, whether in Source or Object
form, that is based on (or derived from) the Work and for which the
editorial revisions, annotations, elaborations, or other modifications
represent, as a whole, an original work of authorship. For the purposes
of this License, Derivative Works shall not include works that remain
separable from, or merely link (or bind by name) to the interfaces of,
the Work and Derivative Works thereof.
4.1 Term. This Agreement shall be effective upon download or install of the
Software.
"Contribution" shall mean any work of authorship, including
the original version of the Work and any modifications or additions
to that Work or Derivative Works thereof, that is intentionally
submitted to Licensor for inclusion in the Work by the copyright owner
or by an individual or Legal Entity authorized to submit on behalf of
the copyright owner. For the purposes of this definition, "submitted"
means any form of electronic, verbal, or written communication sent
to the Licensor or its representatives, including but not limited to
communication on electronic mailing lists, source code control systems,
and issue tracking systems that are managed by, or on behalf of, the
Licensor for the purpose of discussing and improving the Work, but
excluding communication that is conspicuously marked or otherwise
designated in writing by the copyright owner as "Not a Contribution."
4.2 Termination. This Agreement may be terminated by CloudFlare or its
authorized representative by written notice to you if any of the following
events occur: (i) you fail to pay any amounts due for the Services and the
Solution when due and after written notice of such nonpayment has been given to
you; (ii) you are in material breach of any term, condition, or provision of
this Agreement or any other agreement executed by you with CloudFlare or its
authorized representative in connection with the provision of the Solution and
Services (a "Related Agreement"); or (iii) you terminate or suspend your
business, becomes subject to any bankruptcy or insolvency proceeding under
federal or state statutes, or become insolvent or subject to direct control by a
trustee, receiver or similar authority.
"Contributor" shall mean Licensor and any individual or Legal Entity
on behalf of whom a Contribution has been received by Licensor and
subsequently incorporated within the Work.
4.3 Effect of Termination. Upon the termination of this Agreement for any
reason: (1) all license rights granted hereunder shall terminate and (2) all
Confidential Information shall be returned to the disclosing party or destroyed.
2. Grant of Copyright License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
copyright license to reproduce, prepare Derivative Works of,
publicly display, publicly perform, sublicense, and distribute the
Work and such Derivative Works in Source or Object form.
5. MISCELLANEOUS
3. Grant of Patent License. Subject to the terms and conditions of
this License, each Contributor hereby grants to You a perpetual,
worldwide, non-exclusive, no-charge, royalty-free, irrevocable
(except as stated in this section) patent license to make, have made,
use, offer to sell, sell, import, and otherwise transfer the Work,
where such license applies only to those patent claims licensable
by such Contributor that are necessarily infringed by their
Contribution(s) alone or by combination of their Contribution(s)
with the Work to which such Contribution(s) was submitted. If You
institute patent litigation against any entity (including a
cross-claim or counterclaim in a lawsuit) alleging that the Work
or a Contribution incorporated within the Work constitutes direct
or contributory patent infringement, then any patent licenses
granted to You under this License for that Work shall terminate
as of the date such litigation is filed.
5.1 Assignment. You may not assign any of your rights or delegate any of
your obligations under this Agreement, whether by operation of law or otherwise,
without the prior express written consent of CloudFlare or its authorized
representative. Any such assignment without the prior express written consent
of CloudFlare or its authorized representative shall be void. Subject to the
foregoing, this Agreement will bind and inure to the benefit of the parties,
their respective successors and permitted assigns.
4. Redistribution. You may reproduce and distribute copies of the
Work or Derivative Works thereof in any medium, with or without
modifications, and in Source or Object form, provided that You
meet the following conditions:
5.2 Waiver and Amendment. No modification, amendment or waiver of any
provision of this Agreement shall be effective unless in writing and signed by
the party to be charged. No failure or delay by either party in exercising any
right, power, or remedy under this Agreement, except as specifically provided
herein, shall operate as a waiver of any such right, power or remedy. Without
limiting the foregoing, terms and conditions on any purchase orders or similar
materials submitted by you to CloudFlare or its authorized representative shall
be of no force or effect.
(a) You must give any other recipients of the Work or
Derivative Works a copy of this License; and
5.3 Governing Law. This Agreement shall be governed by the laws of the State
of California, USA, excluding conflict of laws and provisions, and excluding the
United Nations Convention on Contracts for the International Sale of Goods.
(b) You must cause any modified files to carry prominent notices
stating that You changed the files; and
5.4 Notices. All notices, demands or consents required or permitted under
this Agreement shall be in writing. Notice shall be sent to you at the e-mail
address provided by you to CloudFlare or its authorized representative in
connection with the Solution.
(c) You must retain, in the Source form of any Derivative Works
that You distribute, all copyright, patent, trademark, and
attribution notices from the Source form of the Work,
excluding those notices that do not pertain to any part of
the Derivative Works; and
5.5 Independent Contractors. The parties are independent contractors.
Neither party shall be deemed to be an employee, agent, partner or legal
representative of the other for any purpose and neither shall have any right,
power or authority to create any obligation or responsibility on behalf of the
other.
(d) If the Work includes a "NOTICE" text file as part of its
distribution, then any Derivative Works that You distribute must
include a readable copy of the attribution notices contained
within such NOTICE file, excluding those notices that do not
pertain to any part of the Derivative Works, in at least one
of the following places: within a NOTICE text file distributed
as part of the Derivative Works; within the Source form or
documentation, if provided along with the Derivative Works; or,
within a display generated by the Derivative Works, if and
wherever such third-party notices normally appear. The contents
of the NOTICE file are for informational purposes only and
do not modify the License. You may add Your own attribution
notices within Derivative Works that You distribute, alongside
or as an addendum to the NOTICE text from the Work, provided
that such additional attribution notices cannot be construed
as modifying the License.
5.6 Severability. If any provision of this Agreement is held by a court of
competent jurisdiction to be contrary to law, such provision shall be changed
and interpreted so as to best accomplish the objectives of the original
provision to the fullest extent allowed by law and the remaining provisions of
this Agreement shall remain in full force and effect.
You may add Your own copyright statement to Your modifications and
may provide additional or different license terms and conditions
for use, reproduction, or distribution of Your modifications, or
for any such Derivative Works as a whole, provided Your use,
reproduction, and distribution of the Work otherwise complies with
the conditions stated in this License.
5.7 Force Majeure. CloudFlare shall not be liable to the other party for any
failure or delay in performance caused by reasons beyond its reasonable control.
5. Submission of Contributions. Unless You explicitly state otherwise,
any Contribution intentionally submitted for inclusion in the Work
by You to the Licensor shall be under the terms and conditions of
this License, without any additional terms or conditions.
Notwithstanding the above, nothing herein shall supersede or modify
the terms of any separate license agreement you may have executed
with Licensor regarding such Contributions.
5.8 Complete Understanding. This Agreement and the Related Agreement
constitute the final, complete and exclusive agreement between the parties with
respect to the subject matter hereof, and supersedes all previous written and
oral agreements and communications related to the subject matter of this
Agreement. To the extent this Agreement and the Related Agreement conflict,
this Agreement shall control.
6. Trademarks. This License does not grant permission to use the trade
names, trademarks, service marks, or product names of the Licensor,
except as required for reasonable and customary use in describing the
origin of the Work and reproducing the content of the NOTICE file.
7. Disclaimer of Warranty. Unless required by applicable law or
agreed to in writing, Licensor provides the Work (and each
Contributor provides its Contributions) on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
implied, including, without limitation, any warranties or conditions
of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
PARTICULAR PURPOSE. You are solely responsible for determining the
appropriateness of using or redistributing the Work and assume any
risks associated with Your exercise of permissions under this License.
8. Limitation of Liability. In no event and under no legal theory,
whether in tort (including negligence), contract, or otherwise,
unless required by applicable law (such as deliberate and grossly
negligent acts) or agreed to in writing, shall any Contributor be
liable to You for damages, including any direct, indirect, special,
incidental, or consequential damages of any character arising as a
result of this License or out of the use or inability to use the
Work (including but not limited to damages for loss of goodwill,
work stoppage, computer failure or malfunction, or any and all
other commercial damages or losses), even if such Contributor
has been advised of the possibility of such damages.
9. Accepting Warranty or Additional Liability. While redistributing
the Work or Derivative Works thereof, You may choose to offer,
and charge a fee for, acceptance of support, warranty, indemnity,
or other liability obligations and/or rights consistent with this
License. However, in accepting such obligations, You may act only
on Your own behalf and on Your sole responsibility, not on behalf
of any other Contributor, and only if You agree to indemnify,
defend, and hold each Contributor harmless for any liability
incurred by, or claims asserted against, such Contributor by reason
of your accepting any such warranty or additional liability.
END OF TERMS AND CONDITIONS
APPENDIX: How to apply the Apache License to your work.
To apply the Apache License to your work, attach the following
boilerplate notice, with the fields enclosed by brackets "[]"
replaced with your own identifying information. (Don't include
the brackets!) The text should be enclosed in the appropriate
comment syntax for the file format. We also recommend that a
file or class name and description of purpose be included on the
same "printed page" as the copyright notice for easier
identification within third-party archives.
Copyright [yyyy] [name of copyright owner]
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
## Runtime Library Exception to the Apache 2.0 License: ##
As an exception, if you use this Software to compile your source code and
portions of this Software are embedded into the binary product as a result,
you may redistribute such product without providing attribution as would
otherwise be required by Sections 4(a), 4(b) and 4(d) of the License.
+17
View File
@@ -1,3 +1,20 @@
2022.2.2
- 2022-02-22 TUN-5754: Allow ingress validate to take plaintext option
- 2022-02-17 TUN-5678: Cloudflared uses typed tunnel API
2022.2.1
- 2022-02-10 TUN-5184: Handle errors in bidrectional streaming (websocket#Stream) gracefully when 1 side has ended
- 2022-02-14 Update issue templates
- 2022-02-14 Update issue templates
- 2022-02-11 TUN-5768: Update cloudflared license file
- 2022-02-11 TUN-5698: Make ingress rules and warp routing dynamically configurable
- 2022-02-14 TUN-5678: Adapt cloudflared to use new typed APIs
- 2022-02-17 Revert "TUN-5678: Adapt cloudflared to use new typed APIs"
- 2022-02-11 TUN-5697: Listen for UpdateConfiguration RPC in quic transport
- 2022-02-04 TUN-5744: Add a test to make sure cloudflared uses scheme defined in ingress rule, not X-Forwarded-Proto header
- 2022-02-07 TUN-5749: Refactor cloudflared to pave way for reconfigurable ingress - Split origin into supervisor and proxy packages - Create configManager to handle dynamic config
- 2021-10-19 TUN-5184: Make sure outstanding websocket write is finished, and no more writes after shutdown
2022.2.0
- 2022-02-02 TUN-4947: Use http when talking to Unix sockets origins
- 2022-02-02 TUN-5695: Define RPC method to update configuration
+1 -1
View File
@@ -48,7 +48,7 @@ func NewRESTClient(baseURL, accountTag, zoneTag, authToken, userAgent string, lo
if strings.HasSuffix(baseURL, "/") {
baseURL = baseURL[:len(baseURL)-1]
}
accountLevelEndpoint, err := url.Parse(fmt.Sprintf("%s/accounts/%s/tunnels", baseURL, accountTag))
accountLevelEndpoint, err := url.Parse(fmt.Sprintf("%s/accounts/%s/cfd_tunnel", baseURL, accountTag))
if err != nil {
return nil, errors.Wrap(err, "failed to create account level endpoint")
}
+11 -9
View File
@@ -31,8 +31,9 @@ import (
"github.com/cloudflare/cloudflared/ingress"
"github.com/cloudflare/cloudflared/logger"
"github.com/cloudflare/cloudflared/metrics"
"github.com/cloudflare/cloudflared/origin"
"github.com/cloudflare/cloudflared/orchestration"
"github.com/cloudflare/cloudflared/signal"
"github.com/cloudflare/cloudflared/supervisor"
"github.com/cloudflare/cloudflared/tlsconfig"
"github.com/cloudflare/cloudflared/tunneldns"
)
@@ -223,7 +224,7 @@ func routeFromFlag(c *cli.Context) (route cfapi.HostnameRoute, ok bool) {
func StartServer(
c *cli.Context,
info *cliutil.BuildInfo,
namedTunnel *connection.NamedTunnelConfig,
namedTunnel *connection.NamedTunnelProperties,
log *zerolog.Logger,
isUIEnabled bool,
) error {
@@ -333,7 +334,7 @@ func StartServer(
observer.SendURL(quickTunnelURL)
}
tunnelConfig, ingressRules, err := prepareTunnelConfig(c, info, log, logTransport, observer, namedTunnel)
tunnelConfig, dynamicConfig, err := prepareTunnelConfig(c, info, log, logTransport, observer, namedTunnel)
if err != nil {
log.Err(err).Msg("Couldn't start tunnel")
return err
@@ -353,11 +354,12 @@ func StartServer(
errC <- metrics.ServeMetrics(metricsListener, ctx.Done(), readinessServer, quickTunnelURL, log)
}()
if err := ingressRules.StartOrigins(&wg, log, ctx.Done(), errC); err != nil {
orchestrator, err := orchestration.NewOrchestrator(ctx, dynamicConfig, tunnelConfig.Tags, tunnelConfig.Log)
if err != nil {
return err
}
reconnectCh := make(chan origin.ReconnectSignal, 1)
reconnectCh := make(chan supervisor.ReconnectSignal, 1)
if c.IsSet("stdin-control") {
log.Info().Msg("Enabling control through stdin")
go stdinControl(reconnectCh, log)
@@ -369,7 +371,7 @@ func StartServer(
wg.Done()
log.Info().Msg("Tunnel server stopped")
}()
errC <- origin.StartTunnelDaemon(ctx, tunnelConfig, connectedSignal, reconnectCh, graceShutdownC)
errC <- supervisor.StartTunnelDaemon(ctx, tunnelConfig, orchestrator, connectedSignal, reconnectCh, graceShutdownC)
}()
if isUIEnabled {
@@ -377,7 +379,7 @@ func StartServer(
info.Version(),
hostname,
metricsListener.Addr().String(),
&ingressRules,
dynamicConfig.Ingress,
tunnelConfig.HAConnections,
)
app := tunnelUI.Launch(ctx, log, logTransport)
@@ -998,7 +1000,7 @@ func configureProxyDNSFlags(shouldHide bool) []cli.Flag {
}
}
func stdinControl(reconnectCh chan origin.ReconnectSignal, log *zerolog.Logger) {
func stdinControl(reconnectCh chan supervisor.ReconnectSignal, log *zerolog.Logger) {
for {
scanner := bufio.NewScanner(os.Stdin)
for scanner.Scan() {
@@ -1009,7 +1011,7 @@ func stdinControl(reconnectCh chan origin.ReconnectSignal, log *zerolog.Logger)
case "":
break
case "reconnect":
var reconnect origin.ReconnectSignal
var reconnect supervisor.ReconnectSignal
if len(parts) > 1 {
var err error
if reconnect.Delay, err = time.ParseDuration(parts[1]); err != nil {
+43 -48
View File
@@ -23,7 +23,8 @@ import (
"github.com/cloudflare/cloudflared/edgediscovery"
"github.com/cloudflare/cloudflared/h2mux"
"github.com/cloudflare/cloudflared/ingress"
"github.com/cloudflare/cloudflared/origin"
"github.com/cloudflare/cloudflared/orchestration"
"github.com/cloudflare/cloudflared/supervisor"
"github.com/cloudflare/cloudflared/tlsconfig"
tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
"github.com/cloudflare/cloudflared/validation"
@@ -87,7 +88,7 @@ func logClientOptions(c *cli.Context, log *zerolog.Logger) {
}
}
func dnsProxyStandAlone(c *cli.Context, namedTunnel *connection.NamedTunnelConfig) bool {
func dnsProxyStandAlone(c *cli.Context, namedTunnel *connection.NamedTunnelProperties) bool {
return c.IsSet("proxy-dns") && (!c.IsSet("hostname") && !c.IsSet("tag") && !c.IsSet("hello-world") && namedTunnel == nil)
}
@@ -152,44 +153,44 @@ func prepareTunnelConfig(
info *cliutil.BuildInfo,
log, logTransport *zerolog.Logger,
observer *connection.Observer,
namedTunnel *connection.NamedTunnelConfig,
) (*origin.TunnelConfig, ingress.Ingress, error) {
namedTunnel *connection.NamedTunnelProperties,
) (*supervisor.TunnelConfig, *orchestration.Config, error) {
isNamedTunnel := namedTunnel != nil
configHostname := c.String("hostname")
hostname, err := validation.ValidateHostname(configHostname)
if err != nil {
log.Err(err).Str(LogFieldHostname, configHostname).Msg("Invalid hostname")
return nil, ingress.Ingress{}, errors.Wrap(err, "Invalid hostname")
return nil, nil, errors.Wrap(err, "Invalid hostname")
}
clientID := c.String("id")
if !c.IsSet("id") {
clientID, err = generateRandomClientID(log)
if err != nil {
return nil, ingress.Ingress{}, err
return nil, nil, err
}
}
tags, err := NewTagSliceFromCLI(c.StringSlice("tag"))
if err != nil {
log.Err(err).Msg("Tag parse failure")
return nil, ingress.Ingress{}, errors.Wrap(err, "Tag parse failure")
return nil, nil, errors.Wrap(err, "Tag parse failure")
}
tags = append(tags, tunnelpogs.Tag{Name: "ID", Value: clientID})
var (
ingressRules ingress.Ingress
classicTunnel *connection.ClassicTunnelConfig
classicTunnel *connection.ClassicTunnelProperties
)
cfg := config.GetConfiguration()
if isNamedTunnel {
clientUUID, err := uuid.NewRandom()
if err != nil {
return nil, ingress.Ingress{}, errors.Wrap(err, "can't generate connector UUID")
return nil, nil, errors.Wrap(err, "can't generate connector UUID")
}
log.Info().Msgf("Generated Connector ID: %s", clientUUID)
features := append(c.StringSlice("features"), origin.FeatureSerializedHeaders)
features := append(c.StringSlice("features"), supervisor.FeatureSerializedHeaders)
namedTunnel.Client = tunnelpogs.ClientInfo{
ClientID: clientUUID[:],
Features: dedup(features),
@@ -198,10 +199,10 @@ func prepareTunnelConfig(
}
ingressRules, err = ingress.ParseIngress(cfg)
if err != nil && err != ingress.ErrNoIngressRules {
return nil, ingress.Ingress{}, err
return nil, nil, err
}
if !ingressRules.IsEmpty() && c.IsSet("url") {
return nil, ingress.Ingress{}, ingress.ErrURLIncompatibleWithIngress
return nil, nil, ingress.ErrURLIncompatibleWithIngress
}
} else {
@@ -212,10 +213,10 @@ func prepareTunnelConfig(
originCert, err := getOriginCert(originCertPath, &originCertLog)
if err != nil {
return nil, ingress.Ingress{}, errors.Wrap(err, "Error getting origin cert")
return nil, nil, errors.Wrap(err, "Error getting origin cert")
}
classicTunnel = &connection.ClassicTunnelConfig{
classicTunnel = &connection.ClassicTunnelProperties{
Hostname: hostname,
OriginCert: originCert,
// turn off use of reconnect token and auth refresh when using named tunnels
@@ -227,20 +228,14 @@ func prepareTunnelConfig(
if ingressRules.IsEmpty() {
ingressRules, err = ingress.NewSingleOrigin(c, !isNamedTunnel)
if err != nil {
return nil, ingress.Ingress{}, err
return nil, nil, err
}
}
var warpRoutingService *ingress.WarpRoutingService
warpRoutingEnabled := isWarpRoutingEnabled(cfg.WarpRouting, isNamedTunnel)
if warpRoutingEnabled {
warpRoutingService = ingress.NewWarpRoutingService()
log.Info().Msgf("Warp-routing is enabled")
}
protocolSelector, err := connection.NewProtocolSelector(c.String("protocol"), warpRoutingEnabled, namedTunnel, edgediscovery.ProtocolPercentage, origin.ResolveTTL, log)
protocolSelector, err := connection.NewProtocolSelector(c.String("protocol"), warpRoutingEnabled, namedTunnel, edgediscovery.ProtocolPercentage, supervisor.ResolveTTL, log)
if err != nil {
return nil, ingress.Ingress{}, err
return nil, nil, err
}
log.Info().Msgf("Initial protocol %s", protocolSelector.Current())
@@ -248,11 +243,11 @@ func prepareTunnelConfig(
for _, p := range connection.ProtocolList {
tlsSettings := p.TLSSettings()
if tlsSettings == nil {
return nil, ingress.Ingress{}, fmt.Errorf("%s has unknown TLS settings", p)
return nil, nil, fmt.Errorf("%s has unknown TLS settings", p)
}
edgeTLSConfig, err := tlsconfig.CreateTunnelConfig(c, tlsSettings.ServerName)
if err != nil {
return nil, ingress.Ingress{}, errors.Wrap(err, "unable to create TLS config to connect with edge")
return nil, nil, errors.Wrap(err, "unable to create TLS config to connect with edge")
}
if len(tlsSettings.NextProtos) > 0 {
edgeTLSConfig.NextProtos = tlsSettings.NextProtos
@@ -260,15 +255,9 @@ func prepareTunnelConfig(
edgeTLSConfigs[p] = edgeTLSConfig
}
originProxy := origin.NewOriginProxy(ingressRules, warpRoutingService, tags, log)
gracePeriod, err := gracePeriod(c)
if err != nil {
return nil, ingress.Ingress{}, err
}
connectionConfig := &connection.Config{
OriginProxy: originProxy,
GracePeriod: gracePeriod,
ReplaceExisting: c.Bool("force"),
return nil, nil, err
}
muxerConfig := &connection.MuxerConfig{
HeartbeatInterval: c.Duration("heartbeat-interval"),
@@ -279,21 +268,22 @@ func prepareTunnelConfig(
MetricsUpdateFreq: c.Duration("metrics-update-freq"),
}
return &origin.TunnelConfig{
ConnectionConfig: connectionConfig,
OSArch: info.OSArch(),
ClientID: clientID,
EdgeAddrs: c.StringSlice("edge"),
Region: c.String("region"),
HAConnections: c.Int("ha-connections"),
IncidentLookup: origin.NewIncidentLookup(),
IsAutoupdated: c.Bool("is-autoupdated"),
LBPool: c.String("lb-pool"),
Tags: tags,
Log: log,
LogTransport: logTransport,
Observer: observer,
ReportedVersion: info.Version(),
tunnelConfig := &supervisor.TunnelConfig{
GracePeriod: gracePeriod,
ReplaceExisting: c.Bool("force"),
OSArch: info.OSArch(),
ClientID: clientID,
EdgeAddrs: c.StringSlice("edge"),
Region: c.String("region"),
HAConnections: c.Int("ha-connections"),
IncidentLookup: supervisor.NewIncidentLookup(),
IsAutoupdated: c.Bool("is-autoupdated"),
LBPool: c.String("lb-pool"),
Tags: tags,
Log: log,
LogTransport: logTransport,
Observer: observer,
ReportedVersion: info.Version(),
// Note TUN-3758 , we use Int because UInt is not supported with altsrc
Retries: uint(c.Int("retries")),
RunFromTerminal: isRunningFromTerminal(),
@@ -302,7 +292,12 @@ func prepareTunnelConfig(
MuxerConfig: muxerConfig,
ProtocolSelector: protocolSelector,
EdgeTLSConfigs: edgeTLSConfigs,
}, ingressRules, nil
}
dynamicConfig := &orchestration.Config{
Ingress: &ingressRules,
WarpRoutingEnabled: warpRoutingEnabled,
}
return tunnelConfig, dynamicConfig, nil
}
func gracePeriod(c *cli.Context) (time.Duration, error) {
+31 -5
View File
@@ -1,6 +1,7 @@
package tunnel
import (
"encoding/json"
"fmt"
"net/url"
@@ -12,6 +13,15 @@ import (
"github.com/urfave/cli/v2"
)
const ingressDataJSONFlagName = "json"
var ingressDataJSON = &cli.StringFlag{
Name: ingressDataJSONFlagName,
Aliases: []string{"j"},
Usage: `Accepts data in the form of json as an input rather than read from a file`,
EnvVars: []string{"TUNNEL_INGRESS_VALIDATE_JSON"},
}
func buildIngressSubcommand() *cli.Command {
return &cli.Command{
Name: "ingress",
@@ -49,6 +59,7 @@ func buildValidateIngressCommand() *cli.Command {
Usage: "Validate the ingress configuration ",
UsageText: "cloudflared tunnel [--config FILEPATH] ingress validate",
Description: "Validates the configuration file, ensuring your ingress rules are OK.",
Flags: []cli.Flag{ingressDataJSON},
}
}
@@ -69,12 +80,11 @@ func buildTestURLCommand() *cli.Command {
// validateIngressCommand check the syntax of the ingress rules in the cloudflared config file
func validateIngressCommand(c *cli.Context, warnings string) error {
conf := config.GetConfiguration()
if conf.Source() == "" {
fmt.Println("No configuration file was found. Please create one, or use the --config flag to specify its filepath. You can use the help command to learn more about configuration files")
return nil
conf, err := getConfiguration(c)
if err != nil {
return err
}
fmt.Println("Validating rules from", conf.Source())
if _, err := ingress.ParseIngress(conf); err != nil {
return errors.Wrap(err, "Validation failed")
}
@@ -90,6 +100,22 @@ func validateIngressCommand(c *cli.Context, warnings string) error {
return nil
}
func getConfiguration(c *cli.Context) (*config.Configuration, error) {
var conf *config.Configuration
if c.IsSet(ingressDataJSONFlagName) {
ingressJSON := c.String(ingressDataJSONFlagName)
fmt.Println("Validating rules from cmdline flag --json")
err := json.Unmarshal([]byte(ingressJSON), &conf)
return conf, err
}
conf = config.GetConfiguration()
if conf.Source() == "" {
return nil, errors.New("No configuration file was found. Please create one, or use the --config flag to specify its filepath. You can use the help command to learn more about configuration files")
}
fmt.Println("Validating rules from", conf.Source())
return conf, nil
}
// testURLCommand checks which ingress rule matches the given URL.
func testURLCommand(c *cli.Context) error {
requestArg := c.Args().First()
+1 -1
View File
@@ -77,7 +77,7 @@ func RunQuickTunnel(sc *subcommandContext) error {
return StartServer(
sc.c,
buildInfo,
&connection.NamedTunnelConfig{Credentials: credentials, QuickTunnelUrl: data.Result.Hostname},
&connection.NamedTunnelProperties{Credentials: credentials, QuickTunnelUrl: data.Result.Hostname},
sc.log,
sc.isUIEnabled,
)
+1 -1
View File
@@ -304,7 +304,7 @@ func (sc *subcommandContext) run(tunnelID uuid.UUID) error {
return StartServer(
sc.c,
buildInfo,
&connection.NamedTunnelConfig{Credentials: credentials},
&connection.NamedTunnelProperties{Credentials: credentials},
sc.log,
sc.isUIEnabled,
)
@@ -11,6 +11,7 @@ import (
"github.com/google/uuid"
"github.com/pkg/errors"
"github.com/rs/zerolog"
"github.com/stretchr/testify/assert"
"github.com/urfave/cli/v2"
"github.com/cloudflare/cloudflared/cfapi"
@@ -322,3 +323,48 @@ func Test_subcommandContext_Delete(t *testing.T) {
})
}
}
func Test_subcommandContext_ValidateIngressCommand(t *testing.T) {
var tests = []struct {
name string
c *cli.Context
wantErr bool
expectedErr error
}{
{
name: "read a valid configuration from data",
c: func() *cli.Context {
data := `{ "warp-routing": {"enabled": true}, "originRequest" : {"connectTimeout": 10}, "ingress" : [ {"hostname": "test", "service": "https://localhost:8000" } , {"service": "http_status:404"} ]}`
flagSet := flag.NewFlagSet("json", flag.PanicOnError)
flagSet.String(ingressDataJSONFlagName, data, "")
c := cli.NewContext(cli.NewApp(), flagSet, nil)
_ = c.Set(ingressDataJSONFlagName, data)
return c
}(),
},
{
name: "read an invalid configuration with multiple mistakes",
c: func() *cli.Context {
data := `{ "ingress" : [ {"hostname": "test", "service": "localhost:8000" } , {"service": "http_status:invalid_status"} ]}`
flagSet := flag.NewFlagSet("json", flag.PanicOnError)
flagSet.String(ingressDataJSONFlagName, data, "")
c := cli.NewContext(cli.NewApp(), flagSet, nil)
_ = c.Set(ingressDataJSONFlagName, data)
return c
}(),
wantErr: true,
expectedErr: errors.New("Validation failed: localhost:8000 is an invalid address, please make sure it has a scheme and a hostname"),
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := validateIngressCommand(tt.c, "")
if tt.wantErr {
assert.Equal(t, tt.expectedErr.Error(), err.Error())
} else {
assert.Nil(t, err)
}
})
}
}
+5 -6
View File
@@ -25,13 +25,12 @@ const (
var switchingProtocolText = fmt.Sprintf("%d %s", http.StatusSwitchingProtocols, http.StatusText(http.StatusSwitchingProtocols))
type Config struct {
OriginProxy OriginProxy
GracePeriod time.Duration
ReplaceExisting bool
type Orchestrator interface {
UpdateConfig(version int32, config []byte) *pogs.UpdateConfigurationResponse
GetOriginProxy() (OriginProxy, error)
}
type NamedTunnelConfig struct {
type NamedTunnelProperties struct {
Credentials Credentials
Client pogs.ClientInfo
QuickTunnelUrl string
@@ -52,7 +51,7 @@ func (c *Credentials) Auth() pogs.TunnelAuth {
}
}
type ClassicTunnelConfig struct {
type ClassicTunnelProperties struct {
Hostname string
OriginCert []byte
// feature-flag to use new edge reconnect tokens
+96 -29
View File
@@ -4,33 +4,28 @@ import (
"context"
"fmt"
"io"
"math/rand"
"net/http"
"net/url"
"testing"
"time"
"github.com/gobwas/ws/wsutil"
"github.com/rs/zerolog"
"github.com/stretchr/testify/assert"
"github.com/cloudflare/cloudflared/ingress"
tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
"github.com/cloudflare/cloudflared/websocket"
)
const (
largeFileSize = 2 * 1024 * 1024
largeFileSize = 2 * 1024 * 1024
testGracePeriod = time.Millisecond * 100
)
var (
unusedWarpRoutingService = (*ingress.WarpRoutingService)(nil)
testConfig = &Config{
OriginProxy: &mockOriginProxy{},
GracePeriod: time.Millisecond * 100,
testOrchestrator = &mockOrchestrator{
originProxy: &mockOriginProxy{},
}
log = zerolog.Nop()
testOriginURL = &url.URL{
Scheme: "https",
Host: "connectiontest.argotunnel.com",
}
testLargeResp = make([]byte, largeFileSize)
)
@@ -42,6 +37,20 @@ type testRequest struct {
isProxyError bool
}
type mockOrchestrator struct {
originProxy OriginProxy
}
func (*mockOrchestrator) UpdateConfig(version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
return &tunnelpogs.UpdateConfigurationResponse{
LastAppliedVersion: version,
}
}
func (mcr *mockOrchestrator) GetOriginProxy() (OriginProxy, error) {
return mcr.originProxy, nil
}
type mockOriginProxy struct{}
func (moc *mockOriginProxy) ProxyHTTP(
@@ -50,7 +59,15 @@ func (moc *mockOriginProxy) ProxyHTTP(
isWebsocket bool,
) error {
if isWebsocket {
return wsEndpoint(w, req)
switch req.URL.Path {
case "/ws/echo":
return wsEchoEndpoint(w, req)
case "/ws/flaky":
return wsFlakyEndpoint(w, req)
default:
originRespEndpoint(w, http.StatusNotFound, []byte("ws endpoint not found"))
return fmt.Errorf("Unknwon websocket endpoint %s", req.URL.Path)
}
}
switch req.URL.Path {
case "/ok":
@@ -78,32 +95,82 @@ func (moc *mockOriginProxy) ProxyTCP(
return nil
}
type nowriter struct {
io.Reader
type echoPipe struct {
reader *io.PipeReader
writer *io.PipeWriter
}
func (nowriter) Write(p []byte) (int, error) {
return 0, fmt.Errorf("Writer not implemented")
func (ep *echoPipe) Read(p []byte) (int, error) {
return ep.reader.Read(p)
}
func wsEndpoint(w ResponseWriter, r *http.Request) error {
func (ep *echoPipe) Write(p []byte) (int, error) {
return ep.writer.Write(p)
}
// A mock origin that echos data by streaming like a tcpOverWSConnection
// https://github.com/cloudflare/cloudflared/blob/master/ingress/origin_connection.go
func wsEchoEndpoint(w ResponseWriter, r *http.Request) error {
resp := &http.Response{
StatusCode: http.StatusSwitchingProtocols,
}
_ = w.WriteRespHeaders(resp.StatusCode, resp.Header)
clientReader := nowriter{r.Body}
if err := w.WriteRespHeaders(resp.StatusCode, resp.Header); err != nil {
return err
}
wsCtx, cancel := context.WithCancel(r.Context())
readPipe, writePipe := io.Pipe()
wsConn := websocket.NewConn(wsCtx, NewHTTPResponseReadWriterAcker(w, r), &log)
go func() {
for {
data, err := wsutil.ReadClientText(clientReader)
if err != nil {
return
}
if err := wsutil.WriteServerText(w, data); err != nil {
return
}
select {
case <-wsCtx.Done():
case <-r.Context().Done():
}
readPipe.Close()
writePipe.Close()
}()
<-r.Context().Done()
originConn := &echoPipe{reader: readPipe, writer: writePipe}
websocket.Stream(wsConn, originConn, &log)
cancel()
wsConn.Close()
return nil
}
type flakyConn struct {
closeAt time.Time
}
func (fc *flakyConn) Read(p []byte) (int, error) {
if time.Now().After(fc.closeAt) {
return 0, io.EOF
}
n := copy(p, "Read from flaky connection")
return n, nil
}
func (fc *flakyConn) Write(p []byte) (int, error) {
if time.Now().After(fc.closeAt) {
return 0, fmt.Errorf("flaky connection closed")
}
return len(p), nil
}
func wsFlakyEndpoint(w ResponseWriter, r *http.Request) error {
resp := &http.Response{
StatusCode: http.StatusSwitchingProtocols,
}
if err := w.WriteRespHeaders(resp.StatusCode, resp.Header); err != nil {
return err
}
wsCtx, cancel := context.WithCancel(r.Context())
wsConn := websocket.NewConn(wsCtx, NewHTTPResponseReadWriterAcker(w, r), &log)
closedAfter := time.Millisecond * time.Duration(rand.Intn(50))
originConn := &flakyConn{closeAt: time.Now().Add(closedAfter)}
websocket.Stream(wsConn, originConn, &log)
cancel()
wsConn.Close()
return nil
}
+12 -12
View File
@@ -16,9 +16,9 @@ type RPCClientFunc func(context.Context, io.ReadWriteCloser, *zerolog.Logger) Na
type controlStream struct {
observer *Observer
connectedFuse ConnectedFuse
namedTunnelConfig *NamedTunnelConfig
connIndex uint8
connectedFuse ConnectedFuse
namedTunnelProperties *NamedTunnelProperties
connIndex uint8
newRPCClientFunc RPCClientFunc
@@ -39,7 +39,7 @@ type ControlStreamHandler interface {
func NewControlStream(
observer *Observer,
connectedFuse ConnectedFuse,
namedTunnelConfig *NamedTunnelConfig,
namedTunnelConfig *NamedTunnelProperties,
connIndex uint8,
newRPCClientFunc RPCClientFunc,
gracefulShutdownC <-chan struct{},
@@ -49,13 +49,13 @@ func NewControlStream(
newRPCClientFunc = newRegistrationRPCClient
}
return &controlStream{
observer: observer,
connectedFuse: connectedFuse,
namedTunnelConfig: namedTunnelConfig,
newRPCClientFunc: newRPCClientFunc,
connIndex: connIndex,
gracefulShutdownC: gracefulShutdownC,
gracePeriod: gracePeriod,
observer: observer,
connectedFuse: connectedFuse,
namedTunnelProperties: namedTunnelConfig,
newRPCClientFunc: newRPCClientFunc,
connIndex: connIndex,
gracefulShutdownC: gracefulShutdownC,
gracePeriod: gracePeriod,
}
}
@@ -66,7 +66,7 @@ func (c *controlStream) ServeControlStream(
) error {
rpcClient := c.newRPCClientFunc(ctx, rw, c.observer.log)
if err := rpcClient.RegisterConnection(ctx, c.namedTunnelConfig, connOptions, c.connIndex, c.observer); err != nil {
if err := rpcClient.RegisterConnection(ctx, c.namedTunnelProperties, connOptions, c.connIndex, c.observer); err != nil {
rpcClient.Close()
return err
}
+17 -8
View File
@@ -22,9 +22,10 @@ const (
)
type h2muxConnection struct {
config *Config
muxerConfig *MuxerConfig
muxer *h2mux.Muxer
orchestrator Orchestrator
gracePeriod time.Duration
muxerConfig *MuxerConfig
muxer *h2mux.Muxer
// connectionID is only used by metrics, and prometheus requires labels to be string
connIndexStr string
connIndex uint8
@@ -60,7 +61,8 @@ func (mc *MuxerConfig) H2MuxerConfig(h h2mux.MuxedStreamHandler, log *zerolog.Lo
// NewTunnelHandler returns a TunnelHandler, origin LAN IP and error
func NewH2muxConnection(
config *Config,
orchestrator Orchestrator,
gracePeriod time.Duration,
muxerConfig *MuxerConfig,
edgeConn net.Conn,
connIndex uint8,
@@ -68,7 +70,8 @@ func NewH2muxConnection(
gracefulShutdownC <-chan struct{},
) (*h2muxConnection, error, bool) {
h := &h2muxConnection{
config: config,
orchestrator: orchestrator,
gracePeriod: gracePeriod,
muxerConfig: muxerConfig,
connIndexStr: uint8ToString(connIndex),
connIndex: connIndex,
@@ -88,7 +91,7 @@ func NewH2muxConnection(
return h, nil, false
}
func (h *h2muxConnection) ServeNamedTunnel(ctx context.Context, namedTunnel *NamedTunnelConfig, connOptions *tunnelpogs.ConnectionOptions, connectedFuse ConnectedFuse) error {
func (h *h2muxConnection) ServeNamedTunnel(ctx context.Context, namedTunnel *NamedTunnelProperties, connOptions *tunnelpogs.ConnectionOptions, connectedFuse ConnectedFuse) error {
errGroup, serveCtx := errgroup.WithContext(ctx)
errGroup.Go(func() error {
return h.serveMuxer(serveCtx)
@@ -117,7 +120,7 @@ func (h *h2muxConnection) ServeNamedTunnel(ctx context.Context, namedTunnel *Nam
return err
}
func (h *h2muxConnection) ServeClassicTunnel(ctx context.Context, classicTunnel *ClassicTunnelConfig, credentialManager CredentialManager, registrationOptions *tunnelpogs.RegistrationOptions, connectedFuse ConnectedFuse) error {
func (h *h2muxConnection) ServeClassicTunnel(ctx context.Context, classicTunnel *ClassicTunnelProperties, credentialManager CredentialManager, registrationOptions *tunnelpogs.RegistrationOptions, connectedFuse ConnectedFuse) error {
errGroup, serveCtx := errgroup.WithContext(ctx)
errGroup.Go(func() error {
return h.serveMuxer(serveCtx)
@@ -224,7 +227,13 @@ func (h *h2muxConnection) ServeStream(stream *h2mux.MuxedStream) error {
sourceConnectionType = TypeWebsocket
}
err := h.config.OriginProxy.ProxyHTTP(respWriter, req, sourceConnectionType == TypeWebsocket)
originProxy, err := h.orchestrator.GetOriginProxy()
if err != nil {
respWriter.WriteErrorResponse()
return err
}
err = originProxy.ProxyHTTP(respWriter, req, sourceConnectionType == TypeWebsocket)
if err != nil {
respWriter.WriteErrorResponse()
}
+4 -4
View File
@@ -48,7 +48,7 @@ func newH2MuxConnection(t require.TestingT) (*h2muxConnection, *h2mux.Muxer) {
}()
var connIndex = uint8(0)
testObserver := NewObserver(&log, &log, false)
h2muxConn, err, _ := NewH2muxConnection(testConfig, testMuxerConfig, originConn, connIndex, testObserver, nil)
h2muxConn, err, _ := NewH2muxConnection(testOrchestrator, testGracePeriod, testMuxerConfig, originConn, connIndex, testObserver, nil)
require.NoError(t, err)
return h2muxConn, <-edgeMuxChan
}
@@ -147,7 +147,7 @@ func TestServeStreamWS(t *testing.T) {
headers := []h2mux.Header{
{
Name: ":path",
Value: "/ws",
Value: "/ws/echo",
},
{
Name: "connection",
@@ -167,10 +167,10 @@ func TestServeStreamWS(t *testing.T) {
assert.True(t, hasHeader(stream, ResponseMetaHeader, responseMetaHeaderOrigin))
data := []byte("test websocket")
err = wsutil.WriteClientText(writePipe, data)
err = wsutil.WriteClientBinary(writePipe, data)
require.NoError(t, err)
respBody, err := wsutil.ReadServerText(stream)
respBody, err := wsutil.ReadServerBinary(stream)
require.NoError(t, err)
require.Equal(t, data, respBody, fmt.Sprintf("Expect %s, got %s", string(data), string(respBody)))
+16 -10
View File
@@ -30,12 +30,12 @@ var errEdgeConnectionClosed = fmt.Errorf("connection with edge closed")
// HTTP2Connection represents a net.Conn that uses HTTP2 frames to proxy traffic from the edge to cloudflared on the
// origin.
type HTTP2Connection struct {
conn net.Conn
server *http2.Server
config *Config
connOptions *tunnelpogs.ConnectionOptions
observer *Observer
connIndex uint8
conn net.Conn
server *http2.Server
orchestrator Orchestrator
connOptions *tunnelpogs.ConnectionOptions
observer *Observer
connIndex uint8
// newRPCClientFunc allows us to mock RPCs during testing
newRPCClientFunc func(context.Context, io.ReadWriteCloser, *zerolog.Logger) NamedTunnelRPCClient
@@ -49,7 +49,7 @@ type HTTP2Connection struct {
// NewHTTP2Connection returns a new instance of HTTP2Connection.
func NewHTTP2Connection(
conn net.Conn,
config *Config,
orchestrator Orchestrator,
connOptions *tunnelpogs.ConnectionOptions,
observer *Observer,
connIndex uint8,
@@ -61,7 +61,7 @@ func NewHTTP2Connection(
server: &http2.Server{
MaxConcurrentStreams: MaxConcurrentStreams,
},
config: config,
orchestrator: orchestrator,
connOptions: connOptions,
observer: observer,
connIndex: connIndex,
@@ -106,6 +106,12 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return
}
originProxy, err := c.orchestrator.GetOriginProxy()
if err != nil {
c.observer.log.Error().Msg(err.Error())
return
}
switch connType {
case TypeControlStream:
if err := c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions); err != nil {
@@ -116,7 +122,7 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
case TypeWebsocket, TypeHTTP:
stripWebsocketUpgradeHeader(r)
if err := c.config.OriginProxy.ProxyHTTP(respWriter, r, connType == TypeWebsocket); err != nil {
if err := originProxy.ProxyHTTP(respWriter, r, connType == TypeWebsocket); err != nil {
err := fmt.Errorf("Failed to proxy HTTP: %w", err)
c.log.Error().Err(err)
respWriter.WriteErrorResponse()
@@ -131,7 +137,7 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
rws := NewHTTPResponseReadWriterAcker(respWriter, r)
if err := c.config.OriginProxy.ProxyTCP(r.Context(), rws, &TCPRequest{
if err := originProxy.ProxyTCP(r.Context(), rws, &TCPRequest{
Dest: host,
CFRay: FindCfRayHeader(r),
LBProbe: IsLBProbeRequest(r),
+94 -19
View File
@@ -2,6 +2,7 @@ package connection
import (
"context"
"errors"
"fmt"
"io"
"io/ioutil"
@@ -27,22 +28,23 @@ var (
)
func newTestHTTP2Connection() (*HTTP2Connection, net.Conn) {
edgeConn, originConn := net.Pipe()
edgeConn, cfdConn := net.Pipe()
var connIndex = uint8(0)
log := zerolog.Nop()
obs := NewObserver(&log, &log, false)
controlStream := NewControlStream(
obs,
mockConnectedFuse{},
&NamedTunnelConfig{},
&NamedTunnelProperties{},
connIndex,
nil,
nil,
1*time.Second,
)
return NewHTTP2Connection(
originConn,
testConfig,
cfdConn,
// OriginProxy is set in testConfigManager
testOrchestrator,
&pogs.ConnectionOptions{},
obs,
connIndex,
@@ -130,7 +132,7 @@ type mockNamedTunnelRPCClient struct {
func (mc mockNamedTunnelRPCClient) RegisterConnection(
c context.Context,
config *NamedTunnelConfig,
properties *NamedTunnelProperties,
options *tunnelpogs.ConnectionOptions,
connIndex uint8,
observer *Observer,
@@ -166,6 +168,8 @@ type wsRespWriter struct {
*httptest.ResponseRecorder
readPipe *io.PipeReader
writePipe *io.PipeWriter
closed bool
panicked bool
}
func newWSRespWriter() *wsRespWriter {
@@ -174,46 +178,59 @@ func newWSRespWriter() *wsRespWriter {
httptest.NewRecorder(),
readPipe,
writePipe,
false,
false,
}
}
type nowriter struct {
io.Reader
}
func (nowriter) Write(_ []byte) (int, error) {
return 0, fmt.Errorf("writer not implemented")
}
func (w *wsRespWriter) RespBody() io.ReadWriter {
return nowriter{w.readPipe}
}
func (w *wsRespWriter) Write(data []byte) (n int, err error) {
if w.closed {
w.panicked = true
return 0, errors.New("wsRespWriter panicked")
}
return w.writePipe.Write(data)
}
func (w *wsRespWriter) close() {
w.closed = true
}
func TestServeWS(t *testing.T) {
http2Conn, _ := newTestHTTP2Connection()
ctx, cancel := context.WithCancel(context.Background())
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
http2Conn.Serve(ctx)
}()
respWriter := newWSRespWriter()
readPipe, writePipe := io.Pipe()
req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/ws", readPipe)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/ws/echo", readPipe)
require.NoError(t, err)
req.Header.Set(InternalUpgradeHeader, WebsocketUpgrade)
wg.Add(1)
serveDone := make(chan struct{})
go func() {
defer wg.Done()
defer close(serveDone)
http2Conn.ServeHTTP(respWriter, req)
respWriter.close()
}()
data := []byte("test websocket")
err = wsutil.WriteClientText(writePipe, data)
err = wsutil.WriteClientBinary(writePipe, data)
require.NoError(t, err)
respBody, err := wsutil.ReadServerText(respWriter.RespBody())
respBody, err := wsutil.ReadServerBinary(respWriter.RespBody())
require.NoError(t, err)
require.Equal(t, data, respBody, fmt.Sprintf("Expect %s, got %s", string(data), string(respBody)))
@@ -223,7 +240,65 @@ func TestServeWS(t *testing.T) {
require.Equal(t, http.StatusOK, resp.StatusCode)
require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(ResponseMetaHeader))
<-serveDone
require.False(t, respWriter.panicked)
}
// TestNoWriteAfterServeHTTPReturns is a regression test of https://jira.cfops.it/browse/TUN-5184
// to make sure we don't write to the ResponseWriter after the ServeHTTP method returns
func TestNoWriteAfterServeHTTPReturns(t *testing.T) {
cfdHTTP2Conn, edgeTCPConn := newTestHTTP2Connection()
ctx, cancel := context.WithCancel(context.Background())
var wg sync.WaitGroup
serverDone := make(chan struct{})
go func() {
defer close(serverDone)
cfdHTTP2Conn.Serve(ctx)
}()
edgeTransport := http2.Transport{}
edgeHTTP2Conn, err := edgeTransport.NewClientConn(edgeTCPConn)
require.NoError(t, err)
message := []byte(t.Name())
for i := 0; i < 100; i++ {
wg.Add(1)
go func() {
defer wg.Done()
readPipe, writePipe := io.Pipe()
reqCtx, reqCancel := context.WithCancel(ctx)
req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, "http://localhost:8080/ws/flaky", readPipe)
require.NoError(t, err)
req.Header.Set(InternalUpgradeHeader, WebsocketUpgrade)
resp, err := edgeHTTP2Conn.RoundTrip(req)
require.NoError(t, err)
// http2RespWriter should rewrite status 101 to 200
require.Equal(t, http.StatusOK, resp.StatusCode)
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case <-reqCtx.Done():
return
default:
}
_ = wsutil.WriteClientBinary(writePipe, message)
}
}()
time.Sleep(time.Millisecond * 100)
reqCancel()
}()
}
wg.Wait()
cancel()
<-serverDone
}
func TestServeControlStream(t *testing.T) {
@@ -238,7 +313,7 @@ func TestServeControlStream(t *testing.T) {
controlStream := NewControlStream(
obs,
mockConnectedFuse{},
&NamedTunnelConfig{},
&NamedTunnelProperties{},
1,
rpcClientFactory.newMockRPCClient,
nil,
@@ -288,7 +363,7 @@ func TestFailRegistration(t *testing.T) {
controlStream := NewControlStream(
obs,
mockConnectedFuse{},
&NamedTunnelConfig{},
&NamedTunnelProperties{},
http2Conn.connIndex,
rpcClientFactory.newMockRPCClient,
nil,
@@ -334,7 +409,7 @@ func TestGracefulShutdownHTTP2(t *testing.T) {
controlStream := NewControlStream(
obs,
mockConnectedFuse{},
&NamedTunnelConfig{},
&NamedTunnelProperties{},
http2Conn.connIndex,
rpcClientFactory.newMockRPCClient,
shutdownC,
+1 -1
View File
@@ -195,7 +195,7 @@ type PercentageFetcher func() (edgediscovery.ProtocolPercents, error)
func NewProtocolSelector(
protocolFlag string,
warpRoutingEnabled bool,
namedTunnel *NamedTunnelConfig,
namedTunnel *NamedTunnelProperties,
fetchFunc PercentageFetcher,
ttl time.Duration,
log *zerolog.Logger,
+23 -23
View File
@@ -16,7 +16,7 @@ const (
)
var (
testNamedTunnelConfig = &NamedTunnelConfig{
testNamedTunnelProperties = &NamedTunnelProperties{
Credentials: Credentials{
AccountTag: "testAccountTag",
},
@@ -51,7 +51,7 @@ func TestNewProtocolSelector(t *testing.T) {
hasFallback bool
expectedFallback Protocol
warpRoutingEnabled bool
namedTunnelConfig *NamedTunnelConfig
namedTunnelConfig *NamedTunnelProperties
fetchFunc PercentageFetcher
wantErr bool
}{
@@ -66,35 +66,35 @@ func TestNewProtocolSelector(t *testing.T) {
protocol: "h2mux",
expectedProtocol: H2mux,
fetchFunc: func() (edgediscovery.ProtocolPercents, error) { return nil, nil },
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "named tunnel over http2",
protocol: "http2",
expectedProtocol: HTTP2,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: 0}),
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "named tunnel http2 disabled still gets http2 because it is manually picked",
protocol: "http2",
expectedProtocol: HTTP2,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: -1}),
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "named tunnel quic disabled still gets quic because it is manually picked",
protocol: "quic",
expectedProtocol: QUIC,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: 100}, edgediscovery.ProtocolPercent{Protocol: "quic", Percentage: -1}),
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "named tunnel quic and http2 disabled",
protocol: "auto",
expectedProtocol: H2mux,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: -1}, edgediscovery.ProtocolPercent{Protocol: "quic", Percentage: -1}),
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "named tunnel quic disabled",
@@ -104,21 +104,21 @@ func TestNewProtocolSelector(t *testing.T) {
hasFallback: true,
expectedFallback: H2mux,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: 100}, edgediscovery.ProtocolPercent{Protocol: "quic", Percentage: -1}),
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "named tunnel auto all http2 disabled",
protocol: "auto",
expectedProtocol: H2mux,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: -1}),
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "named tunnel auto to h2mux",
protocol: "auto",
expectedProtocol: H2mux,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: 0}),
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "named tunnel auto to http2",
@@ -127,7 +127,7 @@ func TestNewProtocolSelector(t *testing.T) {
hasFallback: true,
expectedFallback: H2mux,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: 100}),
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "named tunnel auto to quic",
@@ -136,7 +136,7 @@ func TestNewProtocolSelector(t *testing.T) {
hasFallback: true,
expectedFallback: HTTP2,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "quic", Percentage: 100}),
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "warp routing requesting h2mux",
@@ -145,7 +145,7 @@ func TestNewProtocolSelector(t *testing.T) {
hasFallback: false,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: 100}),
warpRoutingEnabled: true,
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "warp routing requesting h2mux picks HTTP2 even if http2 percent is -1",
@@ -154,7 +154,7 @@ func TestNewProtocolSelector(t *testing.T) {
hasFallback: false,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: -1}),
warpRoutingEnabled: true,
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "warp routing http2",
@@ -163,7 +163,7 @@ func TestNewProtocolSelector(t *testing.T) {
hasFallback: false,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: 100}),
warpRoutingEnabled: true,
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "warp routing quic",
@@ -173,7 +173,7 @@ func TestNewProtocolSelector(t *testing.T) {
expectedFallback: HTTP2Warp,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "quic", Percentage: 100}),
warpRoutingEnabled: true,
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "warp routing auto",
@@ -182,7 +182,7 @@ func TestNewProtocolSelector(t *testing.T) {
hasFallback: false,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: 100}),
warpRoutingEnabled: true,
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
name: "warp routing auto- quic",
@@ -192,7 +192,7 @@ func TestNewProtocolSelector(t *testing.T) {
expectedFallback: HTTP2Warp,
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: 100}, edgediscovery.ProtocolPercent{Protocol: "quic", Percentage: 100}),
warpRoutingEnabled: true,
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
},
{
// None named tunnel can only use h2mux, so specifying an unknown protocol is not an error
@@ -204,14 +204,14 @@ func TestNewProtocolSelector(t *testing.T) {
name: "named tunnel unknown protocol",
protocol: "unknown",
fetchFunc: mockFetcher(false, edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: 100}),
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
wantErr: true,
},
{
name: "named tunnel fetch error",
protocol: "auto",
fetchFunc: mockFetcher(true),
namedTunnelConfig: testNamedTunnelConfig,
namedTunnelConfig: testNamedTunnelProperties,
expectedProtocol: HTTP2,
wantErr: false,
},
@@ -237,7 +237,7 @@ func TestNewProtocolSelector(t *testing.T) {
func TestAutoProtocolSelectorRefresh(t *testing.T) {
fetcher := dynamicMockFetcher{}
selector, err := NewProtocolSelector("auto", noWarpRoutingEnabled, testNamedTunnelConfig, fetcher.fetch(), testNoTTL, &log)
selector, err := NewProtocolSelector("auto", noWarpRoutingEnabled, testNamedTunnelProperties, fetcher.fetch(), testNoTTL, &log)
assert.NoError(t, err)
assert.Equal(t, H2mux, selector.Current())
@@ -267,7 +267,7 @@ func TestAutoProtocolSelectorRefresh(t *testing.T) {
func TestHTTP2ProtocolSelectorRefresh(t *testing.T) {
fetcher := dynamicMockFetcher{}
// Since the user chooses http2 on purpose, we always stick to it.
selector, err := NewProtocolSelector("http2", noWarpRoutingEnabled, testNamedTunnelConfig, fetcher.fetch(), testNoTTL, &log)
selector, err := NewProtocolSelector("http2", noWarpRoutingEnabled, testNamedTunnelProperties, fetcher.fetch(), testNoTTL, &log)
assert.NoError(t, err)
assert.Equal(t, HTTP2, selector.Current())
@@ -297,7 +297,7 @@ func TestHTTP2ProtocolSelectorRefresh(t *testing.T) {
func TestProtocolSelectorRefreshTTL(t *testing.T) {
fetcher := dynamicMockFetcher{}
fetcher.protocolPercents = edgediscovery.ProtocolPercents{edgediscovery.ProtocolPercent{Protocol: "quic", Percentage: 100}}
selector, err := NewProtocolSelector("auto", noWarpRoutingEnabled, testNamedTunnelConfig, fetcher.fetch(), time.Hour, &log)
selector, err := NewProtocolSelector("auto", noWarpRoutingEnabled, testNamedTunnelProperties, fetcher.fetch(), time.Hour, &log)
assert.NoError(t, err)
assert.Equal(t, QUIC, selector.Current())
+11 -7
View File
@@ -36,7 +36,7 @@ const (
type QUICConnection struct {
session quic.Session
logger *zerolog.Logger
httpProxy OriginProxy
orchestrator Orchestrator
sessionManager datagramsession.Manager
controlStreamHandler ControlStreamHandler
connOptions *tunnelpogs.ConnectionOptions
@@ -47,7 +47,7 @@ func NewQUICConnection(
quicConfig *quic.Config,
edgeAddr net.Addr,
tlsConfig *tls.Config,
httpProxy OriginProxy,
orchestrator Orchestrator,
connOptions *tunnelpogs.ConnectionOptions,
controlStreamHandler ControlStreamHandler,
logger *zerolog.Logger,
@@ -66,7 +66,7 @@ func NewQUICConnection(
return &QUICConnection{
session: session,
httpProxy: httpProxy,
orchestrator: orchestrator,
logger: logger,
sessionManager: sessionManager,
controlStreamHandler: controlStreamHandler,
@@ -175,6 +175,10 @@ func (q *QUICConnection) handleDataStream(stream *quicpogs.RequestServerStream)
return err
}
originProxy, err := q.orchestrator.GetOriginProxy()
if err != nil {
return err
}
switch connectRequest.Type {
case quicpogs.ConnectionTypeHTTP, quicpogs.ConnectionTypeWebsocket:
req, err := buildHTTPRequest(connectRequest, stream)
@@ -183,10 +187,10 @@ func (q *QUICConnection) handleDataStream(stream *quicpogs.RequestServerStream)
}
w := newHTTPResponseAdapter(stream)
return q.httpProxy.ProxyHTTP(w, req, connectRequest.Type == quicpogs.ConnectionTypeWebsocket)
return originProxy.ProxyHTTP(w, req, connectRequest.Type == quicpogs.ConnectionTypeWebsocket)
case quicpogs.ConnectionTypeTCP:
rwa := &streamReadWriteAcker{stream}
return q.httpProxy.ProxyTCP(context.Background(), rwa, &TCPRequest{Dest: connectRequest.Dest})
return originProxy.ProxyTCP(context.Background(), rwa, &TCPRequest{Dest: connectRequest.Dest})
}
return nil
}
@@ -261,8 +265,8 @@ func (q *QUICConnection) UnregisterUdpSession(ctx context.Context, sessionID uui
}
// UpdateConfiguration is the RPC method invoked by edge when there is a new configuration
func (q *QUICConnection) UpdateConfiguration(ctx context.Context, version int32, config []byte) (*tunnelpogs.UpdateConfigurationResponse, error) {
return nil, fmt.Errorf("TODO: TUN-5698")
func (q *QUICConnection) UpdateConfiguration(ctx context.Context, version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
return q.orchestrator.UpdateConfig(version, config)
}
// streamReadWriteAcker is a light wrapper over QUIC streams with a callback to send response back to
+5 -6
View File
@@ -47,7 +47,7 @@ func TestQUICServer(t *testing.T) {
// This is simply a sample websocket frame message.
wsBuf := &bytes.Buffer{}
wsutil.WriteClientText(wsBuf, []byte("Hello"))
wsutil.WriteClientBinary(wsBuf, []byte("Hello"))
var tests = []struct {
desc string
@@ -104,7 +104,7 @@ func TestQUICServer(t *testing.T) {
},
{
desc: "test ws proxy",
dest: "/ok",
dest: "/ws/echo",
connectionType: quicpogs.ConnectionTypeWebsocket,
metadata: []quicpogs.Metadata{
{
@@ -125,7 +125,7 @@ func TestQUICServer(t *testing.T) {
},
},
message: wsBuf.Bytes(),
expectedResponse: []byte{0x81, 0x5, 0x48, 0x65, 0x6c, 0x6c, 0x6f},
expectedResponse: []byte{0x82, 0x5, 0x48, 0x65, 0x6c, 0x6c, 0x6f},
},
{
desc: "test tcp proxy",
@@ -233,7 +233,7 @@ func (moc *mockOriginProxyWithRequest) ProxyHTTP(w ResponseWriter, r *http.Reque
}
if isWebsocket {
return wsEndpoint(w, r)
return wsEchoEndpoint(w, r)
}
switch r.URL.Path {
case "/ok":
@@ -627,13 +627,12 @@ func testQUICConnection(udpListenerAddr net.Addr, t *testing.T) *QUICConnection
NextProtos: []string{"argotunnel"},
}
// Start a mock httpProxy
originProxy := &mockOriginProxyWithRequest{}
log := zerolog.New(os.Stdout)
qc, err := NewQUICConnection(
testQUICConfig,
udpListenerAddr,
tlsClientConfig,
originProxy,
&mockOrchestrator{originProxy: &mockOriginProxyWithRequest{}},
&tunnelpogs.ConnectionOptions{},
fakeControlStream{},
&log,
+12 -12
View File
@@ -37,7 +37,7 @@ func NewTunnelServerClient(
}
}
func (tsc *tunnelServerClient) Authenticate(ctx context.Context, classicTunnel *ClassicTunnelConfig, registrationOptions *tunnelpogs.RegistrationOptions) (tunnelpogs.AuthOutcome, error) {
func (tsc *tunnelServerClient) Authenticate(ctx context.Context, classicTunnel *ClassicTunnelProperties, registrationOptions *tunnelpogs.RegistrationOptions) (tunnelpogs.AuthOutcome, error) {
authResp, err := tsc.client.Authenticate(ctx, classicTunnel.OriginCert, classicTunnel.Hostname, registrationOptions)
if err != nil {
return nil, err
@@ -54,7 +54,7 @@ func (tsc *tunnelServerClient) Close() {
type NamedTunnelRPCClient interface {
RegisterConnection(
c context.Context,
config *NamedTunnelConfig,
config *NamedTunnelProperties,
options *tunnelpogs.ConnectionOptions,
connIndex uint8,
observer *Observer,
@@ -86,15 +86,15 @@ func newRegistrationRPCClient(
func (rsc *registrationServerClient) RegisterConnection(
ctx context.Context,
config *NamedTunnelConfig,
properties *NamedTunnelProperties,
options *tunnelpogs.ConnectionOptions,
connIndex uint8,
observer *Observer,
) error {
conn, err := rsc.client.RegisterConnection(
ctx,
config.Credentials.Auth(),
config.Credentials.TunnelID,
properties.Credentials.Auth(),
properties.Credentials.TunnelID,
connIndex,
options,
)
@@ -137,7 +137,7 @@ const (
authenticate rpcName = " authenticate"
)
func (h *h2muxConnection) registerTunnel(ctx context.Context, credentialSetter CredentialManager, classicTunnel *ClassicTunnelConfig, registrationOptions *tunnelpogs.RegistrationOptions) error {
func (h *h2muxConnection) registerTunnel(ctx context.Context, credentialSetter CredentialManager, classicTunnel *ClassicTunnelProperties, registrationOptions *tunnelpogs.RegistrationOptions) error {
h.observer.sendRegisteringEvent(registrationOptions.ConnectionID)
stream, err := h.newRPCStream(ctx, register)
@@ -174,7 +174,7 @@ type CredentialManager interface {
func (h *h2muxConnection) processRegistrationSuccess(
registration *tunnelpogs.TunnelRegistration,
name rpcName,
credentialManager CredentialManager, classicTunnel *ClassicTunnelConfig,
credentialManager CredentialManager, classicTunnel *ClassicTunnelProperties,
) error {
for _, logLine := range registration.LogLines {
h.observer.log.Info().Msg(logLine)
@@ -205,7 +205,7 @@ func (h *h2muxConnection) processRegisterTunnelError(err tunnelpogs.TunnelRegist
}
}
func (h *h2muxConnection) reconnectTunnel(ctx context.Context, credentialManager CredentialManager, classicTunnel *ClassicTunnelConfig, registrationOptions *tunnelpogs.RegistrationOptions) error {
func (h *h2muxConnection) reconnectTunnel(ctx context.Context, credentialManager CredentialManager, classicTunnel *ClassicTunnelProperties, registrationOptions *tunnelpogs.RegistrationOptions) error {
token, err := credentialManager.ReconnectToken()
if err != nil {
return err
@@ -264,7 +264,7 @@ func (h *h2muxConnection) logServerInfo(ctx context.Context, rpcClient *tunnelSe
func (h *h2muxConnection) registerNamedTunnel(
ctx context.Context,
namedTunnel *NamedTunnelConfig,
namedTunnel *NamedTunnelProperties,
connOptions *tunnelpogs.ConnectionOptions,
) error {
stream, err := h.newRPCStream(ctx, register)
@@ -283,7 +283,7 @@ func (h *h2muxConnection) registerNamedTunnel(
func (h *h2muxConnection) unregister(isNamedTunnel bool) {
h.observer.sendUnregisteringEvent(h.connIndex)
unregisterCtx, cancel := context.WithTimeout(context.Background(), h.config.GracePeriod)
unregisterCtx, cancel := context.WithTimeout(context.Background(), h.gracePeriod)
defer cancel()
stream, err := h.newRPCStream(unregisterCtx, unregister)
@@ -296,13 +296,13 @@ func (h *h2muxConnection) unregister(isNamedTunnel bool) {
rpcClient := h.newRPCClientFunc(unregisterCtx, stream, h.observer.log)
defer rpcClient.Close()
rpcClient.GracefulShutdown(unregisterCtx, h.config.GracePeriod)
rpcClient.GracefulShutdown(unregisterCtx, h.gracePeriod)
} else {
rpcClient := NewTunnelServerClient(unregisterCtx, stream, h.observer.log)
defer rpcClient.Close()
// gracePeriod is encoded in int64 using capnproto
_ = rpcClient.client.UnregisterTunnel(unregisterCtx, h.config.GracePeriod.Nanoseconds())
_ = rpcClient.client.UnregisterTunnel(unregisterCtx, h.gracePeriod.Nanoseconds())
}
h.observer.log.Info().Uint8(LogFieldConnIndex, h.connIndex).Msg("Unregistered tunnel connection")
+1 -4
View File
@@ -7,7 +7,6 @@ import (
"regexp"
"strconv"
"strings"
"sync"
"github.com/pkg/errors"
"github.com/rs/zerolog"
@@ -145,13 +144,11 @@ func (ing Ingress) IsSingleRule() bool {
// StartOrigins will start any origin services managed by cloudflared, e.g. proxy servers or Hello World.
func (ing Ingress) StartOrigins(
wg *sync.WaitGroup,
log *zerolog.Logger,
shutdownC <-chan struct{},
errC chan error,
) error {
for _, rule := range ing.Rules {
if err := rule.Service.start(wg, log, shutdownC, errC, rule.Config); err != nil {
if err := rule.Service.start(log, shutdownC, rule.Config); err != nil {
return errors.Wrapf(err, "Error starting local service %s", rule.Service)
}
}
+2 -15
View File
@@ -53,7 +53,7 @@ func (wc *tcpOverWSConnection) Stream(ctx context.Context, tunnelConn io.ReadWri
wc.streamHandler(wsConn, wc.conn, log)
cancel()
// Makes sure wsConn stops sending ping before terminating the stream
wsConn.WaitForShutdown()
wsConn.Close()
}
func (wc *tcpOverWSConnection) Close() {
@@ -73,21 +73,8 @@ func (sp *socksProxyOverWSConnection) Stream(ctx context.Context, tunnelConn io.
socks.StreamNetHandler(wsConn, sp.accessPolicy, log)
cancel()
// Makes sure wsConn stops sending ping before terminating the stream
wsConn.WaitForShutdown()
wsConn.Close()
}
func (sp *socksProxyOverWSConnection) Close() {
}
// wsProxyConnection represents a bidirectional stream for a websocket connection to the origin
type wsProxyConnection struct {
rwc io.ReadWriteCloser
}
func (conn *wsProxyConnection) Stream(ctx context.Context, tunnelConn io.ReadWriter, log *zerolog.Logger) {
websocket.Stream(tunnelConn, conn.rwc, log)
}
func (conn *wsProxyConnection) Close() {
conn.rwc.Close()
}
+40 -4
View File
@@ -8,7 +8,6 @@ import (
"net/http"
"net/http/httptest"
"net/url"
"sync"
"testing"
"github.com/stretchr/testify/assert"
@@ -132,10 +131,8 @@ func TestHTTPServiceHostHeaderOverride(t *testing.T) {
httpService := &httpService{
url: originURL,
}
var wg sync.WaitGroup
shutdownC := make(chan struct{})
errC := make(chan error)
require.NoError(t, httpService.start(&wg, testLogger, shutdownC, errC, cfg))
require.NoError(t, httpService.start(testLogger, shutdownC, cfg))
req, err := http.NewRequest(http.MethodGet, originURL.String(), nil)
require.NoError(t, err)
@@ -147,7 +144,46 @@ func TestHTTPServiceHostHeaderOverride(t *testing.T) {
respBody, err := ioutil.ReadAll(resp.Body)
require.NoError(t, err)
require.Equal(t, respBody, []byte(originURL.Host))
}
// TestHTTPServiceUsesIngressRuleScheme makes sure httpService uses scheme defined in ingress rule and not by eyeball request
func TestHTTPServiceUsesIngressRuleScheme(t *testing.T) {
handler := func(w http.ResponseWriter, r *http.Request) {
require.NotNil(t, r.TLS)
// Echo the X-Forwarded-Proto header for assertions
w.Write([]byte(r.Header.Get("X-Forwarded-Proto")))
}
origin := httptest.NewTLSServer(http.HandlerFunc(handler))
defer origin.Close()
originURL, err := url.Parse(origin.URL)
require.NoError(t, err)
require.Equal(t, "https", originURL.Scheme)
cfg := OriginRequestConfig{
NoTLSVerify: true,
}
httpService := &httpService{
url: originURL,
}
shutdownC := make(chan struct{})
require.NoError(t, httpService.start(testLogger, shutdownC, cfg))
// Tunnel uses scheme defined in the service field of the ingress rule, independent of the X-Forwarded-Proto header
protos := []string{"https", "http", "dne"}
for _, p := range protos {
req, err := http.NewRequest(http.MethodGet, originURL.String(), nil)
require.NoError(t, err)
req.Header.Add("X-Forwarded-Proto", p)
resp, err := httpService.RoundTrip(req)
require.NoError(t, err)
require.Equal(t, http.StatusOK, resp.StatusCode)
respBody, err := ioutil.ReadAll(resp.Body)
require.NoError(t, err)
require.Equal(t, respBody, []byte(p))
}
}
func tcpListenRoutine(listener net.Listener, closeChan chan struct{}) {
+16 -20
View File
@@ -8,7 +8,6 @@ import (
"net"
"net/http"
"net/url"
"sync"
"time"
"github.com/pkg/errors"
@@ -20,13 +19,18 @@ import (
"github.com/cloudflare/cloudflared/tlsconfig"
)
const (
HelloWorldService = "Hello World test origin"
)
// OriginService is something a tunnel can proxy traffic to.
type OriginService interface {
String() string
// Start the origin service if it's managed by cloudflared, e.g. proxy servers or Hello World.
// If it's not managed by cloudflared, this is a no-op because the user is responsible for
// starting the origin service.
start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error
// Implementor of services managed by cloudflared should terminate the service if shutdownC is closed
start(log *zerolog.Logger, shutdownC <-chan struct{}, cfg OriginRequestConfig) error
}
// unixSocketPath is an OriginService representing a unix socket (which accepts HTTP)
@@ -39,7 +43,7 @@ func (o *unixSocketPath) String() string {
return "unix socket: " + o.path
}
func (o *unixSocketPath) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
func (o *unixSocketPath) start(log *zerolog.Logger, _ <-chan struct{}, cfg OriginRequestConfig) error {
transport, err := newHTTPTransport(o, cfg, log)
if err != nil {
return err
@@ -54,7 +58,7 @@ type httpService struct {
transport *http.Transport
}
func (o *httpService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
func (o *httpService) start(log *zerolog.Logger, _ <-chan struct{}, cfg OriginRequestConfig) error {
transport, err := newHTTPTransport(o, cfg, log)
if err != nil {
return err
@@ -78,7 +82,7 @@ func (o *rawTCPService) String() string {
return o.name
}
func (o *rawTCPService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
func (o *rawTCPService) start(log *zerolog.Logger, _ <-chan struct{}, cfg OriginRequestConfig) error {
return nil
}
@@ -139,7 +143,7 @@ func (o *tcpOverWSService) String() string {
return o.dest
}
func (o *tcpOverWSService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
func (o *tcpOverWSService) start(log *zerolog.Logger, _ <-chan struct{}, cfg OriginRequestConfig) error {
if cfg.ProxyType == socksProxy {
o.streamHandler = socks.StreamHandler
} else {
@@ -148,7 +152,7 @@ func (o *tcpOverWSService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdo
return nil
}
func (o *socksProxyOverWSService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
func (o *socksProxyOverWSService) start(log *zerolog.Logger, _ <-chan struct{}, cfg OriginRequestConfig) error {
return nil
}
@@ -164,18 +168,16 @@ type helloWorld struct {
}
func (o *helloWorld) String() string {
return "Hello World test origin"
return HelloWorldService
}
// Start starts a HelloWorld server and stores its address in the Service receiver.
func (o *helloWorld) start(
wg *sync.WaitGroup,
log *zerolog.Logger,
shutdownC <-chan struct{},
errC chan error,
cfg OriginRequestConfig,
) error {
if err := o.httpService.start(wg, log, shutdownC, errC, cfg); err != nil {
if err := o.httpService.start(log, shutdownC, cfg); err != nil {
return err
}
@@ -183,11 +185,7 @@ func (o *helloWorld) start(
if err != nil {
return errors.Wrap(err, "Cannot start Hello World Server")
}
wg.Add(1)
go func() {
defer wg.Done()
_ = hello.StartHelloWorldServer(log, helloListener, shutdownC)
}()
go hello.StartHelloWorldServer(log, helloListener, shutdownC)
o.server = helloListener
o.httpService.url = &url.URL{
@@ -218,10 +216,8 @@ func (o *statusCode) String() string {
}
func (o *statusCode) start(
wg *sync.WaitGroup,
log *zerolog.Logger,
shutdownC <-chan struct{},
errC chan error,
_ <-chan struct{},
cfg OriginRequestConfig,
) error {
return nil
@@ -296,6 +292,6 @@ func (mos MockOriginHTTPService) String() string {
return "MockOriginService"
}
func (mos MockOriginHTTPService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
func (mos MockOriginHTTPService) start(log *zerolog.Logger, _ <-chan struct{}, cfg OriginRequestConfig) error {
return nil
}
+15
View File
@@ -0,0 +1,15 @@
package orchestration
import (
"github.com/cloudflare/cloudflared/ingress"
)
type newConfig struct {
ingress.RemoteConfig
// Add more fields when we support other settings in tunnel orchestration
}
type Config struct {
Ingress *ingress.Ingress
WarpRoutingEnabled bool
}
+158
View File
@@ -0,0 +1,158 @@
package orchestration
import (
"context"
"encoding/json"
"fmt"
"sync"
"sync/atomic"
"github.com/pkg/errors"
"github.com/rs/zerolog"
"github.com/cloudflare/cloudflared/connection"
"github.com/cloudflare/cloudflared/ingress"
"github.com/cloudflare/cloudflared/proxy"
tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
)
// Orchestrator manages configurations so they can be updatable during runtime
// properties are static, so it can be read without lock
// currentVersion and config are read/write infrequently, so their access are synchronized with RWMutex
// access to proxy is synchronized with atmoic.Value, because it uses copy-on-write to provide scalable frequently
// read when update is infrequent
type Orchestrator struct {
currentVersion int32
// Used by UpdateConfig to make sure one update at a time
lock sync.RWMutex
// Underlying value is proxy.Proxy, can be read without the lock, but still needs the lock to update
proxy atomic.Value
config *Config
tags []tunnelpogs.Tag
log *zerolog.Logger
// orchestrator must not handle any more updates after shutdownC is closed
shutdownC <-chan struct{}
// Closing proxyShutdownC will close the previous proxy
proxyShutdownC chan<- struct{}
}
func NewOrchestrator(ctx context.Context, config *Config, tags []tunnelpogs.Tag, log *zerolog.Logger) (*Orchestrator, error) {
o := &Orchestrator{
// Lowest possible version, any remote configuration will have version higher than this
currentVersion: 0,
config: config,
tags: tags,
log: log,
shutdownC: ctx.Done(),
}
if err := o.updateIngress(*config.Ingress, config.WarpRoutingEnabled); err != nil {
return nil, err
}
go o.waitToCloseLastProxy()
return o, nil
}
// Update creates a new proxy with the new ingress rules
func (o *Orchestrator) UpdateConfig(version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
o.lock.Lock()
defer o.lock.Unlock()
if o.currentVersion >= version {
o.log.Debug().
Int32("current_version", o.currentVersion).
Int32("received_version", version).
Msg("Current version is equal or newer than receivied version")
return &tunnelpogs.UpdateConfigurationResponse{
LastAppliedVersion: o.currentVersion,
}
}
var newConf newConfig
if err := json.Unmarshal(config, &newConf); err != nil {
o.log.Err(err).
Int32("version", version).
Str("config", string(config)).
Msgf("Failed to deserialize new configuration")
return &tunnelpogs.UpdateConfigurationResponse{
LastAppliedVersion: o.currentVersion,
Err: err,
}
}
if err := o.updateIngress(newConf.Ingress, newConf.WarpRouting.Enabled); err != nil {
o.log.Err(err).
Int32("version", version).
Str("config", string(config)).
Msgf("Failed to update ingress")
return &tunnelpogs.UpdateConfigurationResponse{
LastAppliedVersion: o.currentVersion,
Err: err,
}
}
o.currentVersion = version
o.log.Info().
Int32("version", version).
Str("config", string(config)).
Msg("Updated to new configuration")
return &tunnelpogs.UpdateConfigurationResponse{
LastAppliedVersion: o.currentVersion,
}
}
// The caller is responsible to make sure there is no concurrent access
func (o *Orchestrator) updateIngress(ingressRules ingress.Ingress, warpRoutingEnabled bool) error {
select {
case <-o.shutdownC:
return fmt.Errorf("cloudflared already shutdown")
default:
}
// Start new proxy before closing the ones from last version.
// The upside is we don't need to restart proxy from last version, which can fail
// The downside is new version might have ingress rule that require previous version to be shutdown first
// The downside is minimized because none of the ingress.OriginService implementation have that requirement
proxyShutdownC := make(chan struct{})
if err := ingressRules.StartOrigins(o.log, proxyShutdownC); err != nil {
return errors.Wrap(err, "failed to start origin")
}
newProxy := proxy.NewOriginProxy(ingressRules, warpRoutingEnabled, o.tags, o.log)
o.proxy.Store(newProxy)
o.config.Ingress = &ingressRules
o.config.WarpRoutingEnabled = warpRoutingEnabled
// If proxyShutdownC is nil, there is no previous running proxy
if o.proxyShutdownC != nil {
close(o.proxyShutdownC)
}
o.proxyShutdownC = proxyShutdownC
return nil
}
// GetOriginProxy returns an interface to proxy to origin. It satisfies connection.ConfigManager interface
func (o *Orchestrator) GetOriginProxy() (connection.OriginProxy, error) {
val := o.proxy.Load()
if val == nil {
err := fmt.Errorf("origin proxy not configured")
o.log.Error().Msg(err.Error())
return nil, err
}
proxy, ok := val.(*proxy.Proxy)
if !ok {
err := fmt.Errorf("origin proxy has unexpected value %+v", val)
o.log.Error().Msg(err.Error())
return nil, err
}
return proxy, nil
}
func (o *Orchestrator) waitToCloseLastProxy() {
<-o.shutdownC
o.lock.Lock()
defer o.lock.Unlock()
if o.proxyShutdownC != nil {
close(o.proxyShutdownC)
o.proxyShutdownC = nil
}
}
+686
View File
@@ -0,0 +1,686 @@
package orchestration
import (
"context"
"fmt"
"io"
"io/ioutil"
"net"
"net/http"
"net/http/httptest"
"sync"
"testing"
"time"
"github.com/gobwas/ws/wsutil"
gows "github.com/gorilla/websocket"
"github.com/rs/zerolog"
"github.com/stretchr/testify/require"
"github.com/cloudflare/cloudflared/connection"
"github.com/cloudflare/cloudflared/ingress"
"github.com/cloudflare/cloudflared/proxy"
tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
)
var (
testLogger = zerolog.Logger{}
testTags = []tunnelpogs.Tag{
{
Name: "package",
Value: "orchestration",
},
{
Name: "purpose",
Value: "test",
},
}
)
// TestUpdateConfiguration tests that
// - configurations can be deserialized
// - proxy can be updated
// - last applied version and error are returned
// - configurations can be deserialized
// - receiving an old version is noop
func TestUpdateConfiguration(t *testing.T) {
initConfig := &Config{
Ingress: &ingress.Ingress{},
WarpRoutingEnabled: false,
}
orchestrator, err := NewOrchestrator(context.Background(), initConfig, testTags, &testLogger)
require.NoError(t, err)
initOriginProxy, err := orchestrator.GetOriginProxy()
require.NoError(t, err)
require.IsType(t, &proxy.Proxy{}, initOriginProxy)
configJSONV2 := []byte(`
{
"unknown_field": "not_deserialized",
"originRequest": {
"connectTimeout": 90000000000,
"noHappyEyeballs": true
},
"ingress": [
{
"hostname": "jira.tunnel.org",
"path": "^\/login",
"service": "http://192.16.19.1:443",
"originRequest": {
"noTLSVerify": true,
"connectTimeout": 10000000000
}
},
{
"hostname": "jira.tunnel.org",
"service": "http://172.32.20.6:80",
"originRequest": {
"noTLSVerify": true,
"connectTimeout": 30000000000
}
},
{
"service": "http_status:404"
}
],
"warp-routing": {
"enabled": true
}
}
`)
updateWithValidation(t, orchestrator, 2, configJSONV2)
configV2 := orchestrator.config
// Validate ingress rule 0
require.Equal(t, "jira.tunnel.org", configV2.Ingress.Rules[0].Hostname)
require.True(t, configV2.Ingress.Rules[0].Matches("jira.tunnel.org", "/login"))
require.True(t, configV2.Ingress.Rules[0].Matches("jira.tunnel.org", "/login/2fa"))
require.False(t, configV2.Ingress.Rules[0].Matches("jira.tunnel.org", "/users"))
require.Equal(t, "http://192.16.19.1:443", configV2.Ingress.Rules[0].Service.String())
require.Len(t, configV2.Ingress.Rules, 3)
// originRequest of this ingress rule overrides global default
require.Equal(t, time.Second*10, configV2.Ingress.Rules[0].Config.ConnectTimeout)
require.Equal(t, true, configV2.Ingress.Rules[0].Config.NoTLSVerify)
// Inherited from global default
require.Equal(t, true, configV2.Ingress.Rules[0].Config.NoHappyEyeballs)
// Validate ingress rule 1
require.Equal(t, "jira.tunnel.org", configV2.Ingress.Rules[1].Hostname)
require.True(t, configV2.Ingress.Rules[1].Matches("jira.tunnel.org", "/users"))
require.Equal(t, "http://172.32.20.6:80", configV2.Ingress.Rules[1].Service.String())
// originRequest of this ingress rule overrides global default
require.Equal(t, time.Second*30, configV2.Ingress.Rules[1].Config.ConnectTimeout)
require.Equal(t, true, configV2.Ingress.Rules[1].Config.NoTLSVerify)
// Inherited from global default
require.Equal(t, true, configV2.Ingress.Rules[1].Config.NoHappyEyeballs)
// Validate ingress rule 2, it's the catch-all rule
require.True(t, configV2.Ingress.Rules[2].Matches("blogs.tunnel.io", "/2022/02/10"))
// Inherited from global default
require.Equal(t, time.Second*90, configV2.Ingress.Rules[2].Config.ConnectTimeout)
require.Equal(t, false, configV2.Ingress.Rules[2].Config.NoTLSVerify)
require.Equal(t, true, configV2.Ingress.Rules[2].Config.NoHappyEyeballs)
require.True(t, configV2.WarpRoutingEnabled)
originProxyV2, err := orchestrator.GetOriginProxy()
require.NoError(t, err)
require.IsType(t, &proxy.Proxy{}, originProxyV2)
require.NotEqual(t, originProxyV2, initOriginProxy)
// Should not downgrade to an older version
resp := orchestrator.UpdateConfig(1, nil)
require.NoError(t, resp.Err)
require.Equal(t, int32(2), resp.LastAppliedVersion)
invalidJSON := []byte(`
{
"originRequest":
}
`)
resp = orchestrator.UpdateConfig(3, invalidJSON)
require.Error(t, resp.Err)
require.Equal(t, int32(2), resp.LastAppliedVersion)
originProxyV3, err := orchestrator.GetOriginProxy()
require.NoError(t, err)
require.Equal(t, originProxyV2, originProxyV3)
configJSONV10 := []byte(`
{
"ingress": [
{
"service": "hello-world"
}
],
"warp-routing": {
"enabled": false
}
}
`)
updateWithValidation(t, orchestrator, 10, configJSONV10)
configV10 := orchestrator.config
require.Len(t, configV10.Ingress.Rules, 1)
require.True(t, configV10.Ingress.Rules[0].Matches("blogs.tunnel.io", "/2022/02/10"))
require.Equal(t, ingress.HelloWorldService, configV10.Ingress.Rules[0].Service.String())
require.False(t, configV10.WarpRoutingEnabled)
originProxyV10, err := orchestrator.GetOriginProxy()
require.NoError(t, err)
require.IsType(t, &proxy.Proxy{}, originProxyV10)
require.NotEqual(t, originProxyV10, originProxyV2)
}
// TestConcurrentUpdateAndRead makes sure orchestrator can receive updates and return origin proxy concurrently
func TestConcurrentUpdateAndRead(t *testing.T) {
const (
concurrentRequests = 200
hostname = "public.tunnels.org"
expectedHost = "internal.tunnels.svc.cluster.local"
tcpBody = "testProxyTCP"
)
httpOrigin := httptest.NewServer(&validateHostHandler{
expectedHost: expectedHost,
body: t.Name(),
})
defer httpOrigin.Close()
tcpOrigin, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
defer tcpOrigin.Close()
var (
configJSONV1 = []byte(fmt.Sprintf(`
{
"originRequest": {
"connectTimeout": 90000000000,
"noHappyEyeballs": true
},
"ingress": [
{
"hostname": "%s",
"service": "%s",
"originRequest": {
"httpHostHeader": "%s",
"connectTimeout": 10000000000
}
},
{
"service": "http_status:404"
}
],
"warp-routing": {
"enabled": true
}
}
`, hostname, httpOrigin.URL, expectedHost))
configJSONV2 = []byte(`
{
"ingress": [
{
"service": "http_status:204"
}
],
"warp-routing": {
"enabled": false
}
}
`)
configJSONV3 = []byte(`
{
"ingress": [
{
"service": "http_status:418"
}
],
"warp-routing": {
"enabled": true
}
}
`)
// appliedV2 makes sure v3 is applied after v2
appliedV2 = make(chan struct{})
initConfig = &Config{
Ingress: &ingress.Ingress{},
WarpRoutingEnabled: false,
}
)
orchestrator, err := NewOrchestrator(context.Background(), initConfig, testTags, &testLogger)
require.NoError(t, err)
updateWithValidation(t, orchestrator, 1, configJSONV1)
var wg sync.WaitGroup
// tcpOrigin will be closed when the test exits. Only the handler routines are included in the wait group
go func() {
serveTCPOrigin(t, tcpOrigin, &wg)
}()
for i := 0; i < concurrentRequests; i++ {
originProxy, err := orchestrator.GetOriginProxy()
require.NoError(t, err)
wg.Add(1)
go func(i int, originProxy connection.OriginProxy) {
defer wg.Done()
resp, err := proxyHTTP(t, originProxy, hostname)
require.NoError(t, err)
var warpRoutingDisabled bool
// The response can be from initOrigin, http_status:204 or http_status:418
switch resp.StatusCode {
// v1 proxy, warp enabled
case 200:
body, err := ioutil.ReadAll(resp.Body)
require.NoError(t, err)
require.Equal(t, t.Name(), string(body))
warpRoutingDisabled = false
// v2 proxy, warp disabled
case 204:
require.Greater(t, i, concurrentRequests/4)
warpRoutingDisabled = true
// v3 proxy, warp enabled
case 418:
require.Greater(t, i, concurrentRequests/2)
warpRoutingDisabled = false
}
// Once we have originProxy, it won't be changed by configuration updates.
// We can infer the version by the ProxyHTTP response code
pr, pw := io.Pipe()
// concurrentRespWriter makes sure ResponseRecorder is not read/write concurrently, and read waits for the first write
w := newRespReadWriteFlusher()
// Write TCP message and make sure it's echo back. This has to be done in a go routune since ProxyTCP doesn't
// return until the stream is closed.
if !warpRoutingDisabled {
wg.Add(1)
go func() {
defer wg.Done()
defer pw.Close()
tcpEyeball(t, pw, tcpBody, w)
}()
}
proxyTCP(t, originProxy, tcpOrigin.Addr().String(), w, pr, warpRoutingDisabled)
}(i, originProxy)
if i == concurrentRequests/4 {
wg.Add(1)
go func() {
defer wg.Done()
updateWithValidation(t, orchestrator, 2, configJSONV2)
close(appliedV2)
}()
}
if i == concurrentRequests/2 {
wg.Add(1)
go func() {
defer wg.Done()
<-appliedV2
updateWithValidation(t, orchestrator, 3, configJSONV3)
}()
}
}
wg.Wait()
}
func proxyHTTP(t *testing.T, originProxy connection.OriginProxy, hostname string) (*http.Response, error) {
req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://%s", hostname), nil)
require.NoError(t, err)
w := httptest.NewRecorder()
respWriter, err := connection.NewHTTP2RespWriter(req, w, connection.TypeHTTP)
require.NoError(t, err)
err = originProxy.ProxyHTTP(respWriter, req, false)
if err != nil {
return nil, err
}
return w.Result(), nil
}
func tcpEyeball(t *testing.T, reqWriter io.WriteCloser, body string, respReadWriter *respReadWriteFlusher) {
writeN, err := reqWriter.Write([]byte(body))
require.NoError(t, err)
readBuffer := make([]byte, writeN)
n, err := respReadWriter.Read(readBuffer)
require.NoError(t, err)
require.Equal(t, body, string(readBuffer[:n]))
require.Equal(t, writeN, n)
}
func proxyTCP(t *testing.T, originProxy connection.OriginProxy, originAddr string, w http.ResponseWriter, reqBody io.ReadCloser, expectErr bool) {
req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://%s", originAddr), reqBody)
require.NoError(t, err)
respWriter, err := connection.NewHTTP2RespWriter(req, w, connection.TypeTCP)
require.NoError(t, err)
tcpReq := &connection.TCPRequest{
Dest: originAddr,
CFRay: "123",
LBProbe: false,
}
rws := connection.NewHTTPResponseReadWriterAcker(respWriter, req)
if expectErr {
require.Error(t, originProxy.ProxyTCP(context.Background(), rws, tcpReq))
return
}
require.NoError(t, originProxy.ProxyTCP(context.Background(), rws, tcpReq))
}
func serveTCPOrigin(t *testing.T, tcpOrigin net.Listener, wg *sync.WaitGroup) {
for {
conn, err := tcpOrigin.Accept()
if err != nil {
return
}
wg.Add(1)
go func() {
defer wg.Done()
defer conn.Close()
echoTCP(t, conn)
}()
}
}
func echoTCP(t *testing.T, conn net.Conn) {
readBuf := make([]byte, 1000)
readN, err := conn.Read(readBuf)
require.NoError(t, err)
writeN, err := conn.Write(readBuf[:readN])
require.NoError(t, err)
require.Equal(t, readN, writeN)
}
type validateHostHandler struct {
expectedHost string
body string
}
func (vhh *validateHostHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Host != vhh.expectedHost {
w.WriteHeader(http.StatusBadRequest)
return
}
w.WriteHeader(http.StatusOK)
w.Write([]byte(vhh.body))
}
func updateWithValidation(t *testing.T, orchestrator *Orchestrator, version int32, config []byte) {
resp := orchestrator.UpdateConfig(version, config)
require.NoError(t, resp.Err)
require.Equal(t, version, resp.LastAppliedVersion)
}
// TestClosePreviousProxies makes sure proxies started in the pervious configuration version are shutdown
func TestClosePreviousProxies(t *testing.T) {
var (
hostname = "hello.tunnel1.org"
configWithHelloWorld = []byte(fmt.Sprintf(`
{
"ingress": [
{
"hostname": "%s",
"service": "hello-world"
},
{
"service": "http_status:404"
}
],
"warp-routing": {
"enabled": true
}
}
`, hostname))
configTeapot = []byte(`
{
"ingress": [
{
"service": "http_status:418"
}
],
"warp-routing": {
"enabled": true
}
}
`)
initConfig = &Config{
Ingress: &ingress.Ingress{},
WarpRoutingEnabled: false,
}
)
ctx, cancel := context.WithCancel(context.Background())
orchestrator, err := NewOrchestrator(ctx, initConfig, testTags, &testLogger)
require.NoError(t, err)
updateWithValidation(t, orchestrator, 1, configWithHelloWorld)
originProxyV1, err := orchestrator.GetOriginProxy()
require.NoError(t, err)
resp, err := proxyHTTP(t, originProxyV1, hostname)
require.NoError(t, err)
require.Equal(t, http.StatusOK, resp.StatusCode)
updateWithValidation(t, orchestrator, 2, configTeapot)
originProxyV2, err := orchestrator.GetOriginProxy()
require.NoError(t, err)
resp, err = proxyHTTP(t, originProxyV2, hostname)
require.NoError(t, err)
require.Equal(t, http.StatusTeapot, resp.StatusCode)
// The hello-world server in config v1 should have been stopped
resp, err = proxyHTTP(t, originProxyV1, hostname)
require.Error(t, err)
require.Nil(t, resp)
// Apply the config with hello world server again, orchestrator should spin up another hello world server
updateWithValidation(t, orchestrator, 3, configWithHelloWorld)
originProxyV3, err := orchestrator.GetOriginProxy()
require.NoError(t, err)
require.NotEqual(t, originProxyV1, originProxyV3)
resp, err = proxyHTTP(t, originProxyV3, hostname)
require.NoError(t, err)
require.Equal(t, http.StatusOK, resp.StatusCode)
// cancel the context should terminate the last proxy
cancel()
// Wait for proxies to shutdown
time.Sleep(time.Millisecond * 10)
resp, err = proxyHTTP(t, originProxyV3, hostname)
require.Error(t, err)
require.Nil(t, resp)
}
// TestPersistentConnection makes sure updating the ingress doesn't intefere with existing connections
func TestPersistentConnection(t *testing.T) {
const (
hostname = "http://ws.tunnel.org"
)
msg := t.Name()
initConfig := &Config{
Ingress: &ingress.Ingress{},
WarpRoutingEnabled: false,
}
orchestrator, err := NewOrchestrator(context.Background(), initConfig, testTags, &testLogger)
require.NoError(t, err)
wsOrigin := httptest.NewServer(http.HandlerFunc(wsEcho))
defer wsOrigin.Close()
tcpOrigin, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
defer tcpOrigin.Close()
configWithWSAndWarp := []byte(fmt.Sprintf(`
{
"ingress": [
{
"service": "%s"
}
],
"warp-routing": {
"enabled": true
}
}
`, wsOrigin.URL))
updateWithValidation(t, orchestrator, 1, configWithWSAndWarp)
originProxy, err := orchestrator.GetOriginProxy()
require.NoError(t, err)
wsReqReader, wsReqWriter := io.Pipe()
wsRespReadWriter := newRespReadWriteFlusher()
tcpReqReader, tcpReqWriter := io.Pipe()
tcpRespReadWriter := newRespReadWriteFlusher()
var wg sync.WaitGroup
wg.Add(3)
// Start TCP origin
go func() {
defer wg.Done()
conn, err := tcpOrigin.Accept()
require.NoError(t, err)
defer conn.Close()
// Expect 3 TCP messages
for i := 0; i < 3; i++ {
echoTCP(t, conn)
}
}()
// Simulate cloudflared recieving a TCP connection
go func() {
defer wg.Done()
proxyTCP(t, originProxy, tcpOrigin.Addr().String(), tcpRespReadWriter, tcpReqReader, false)
}()
// Simulate cloudflared recieving a WS connection
go func() {
defer wg.Done()
req, err := http.NewRequest(http.MethodGet, hostname, wsReqReader)
require.NoError(t, err)
// ProxyHTTP will add Connection, Upgrade and Sec-Websocket-Version headers
req.Header.Add("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
respWriter, err := connection.NewHTTP2RespWriter(req, wsRespReadWriter, connection.TypeWebsocket)
require.NoError(t, err)
err = originProxy.ProxyHTTP(respWriter, req, true)
require.NoError(t, err)
}()
// Simulate eyeball WS and TCP connections
validateWsEcho(t, msg, wsReqWriter, wsRespReadWriter)
tcpEyeball(t, tcpReqWriter, msg, tcpRespReadWriter)
configNoWSAndWarp := []byte(`
{
"ingress": [
{
"service": "http_status:404"
}
],
"warp-routing": {
"enabled": false
}
}
`)
updateWithValidation(t, orchestrator, 2, configNoWSAndWarp)
// Make sure connection is still up
validateWsEcho(t, msg, wsReqWriter, wsRespReadWriter)
tcpEyeball(t, tcpReqWriter, msg, tcpRespReadWriter)
updateWithValidation(t, orchestrator, 3, configWithWSAndWarp)
// Make sure connection is still up
validateWsEcho(t, msg, wsReqWriter, wsRespReadWriter)
tcpEyeball(t, tcpReqWriter, msg, tcpRespReadWriter)
wsReqWriter.Close()
tcpReqWriter.Close()
wg.Wait()
}
func wsEcho(w http.ResponseWriter, r *http.Request) {
upgrader := gows.Upgrader{}
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
return
}
defer conn.Close()
for {
mt, message, err := conn.ReadMessage()
if err != nil {
fmt.Println("read message err", err)
break
}
err = conn.WriteMessage(mt, message)
if err != nil {
fmt.Println("write message err", err)
break
}
}
}
func validateWsEcho(t *testing.T, msg string, reqWriter io.Writer, respReadWriter io.ReadWriter) {
err := wsutil.WriteClientText(reqWriter, []byte(msg))
require.NoError(t, err)
receivedMsg, err := wsutil.ReadServerText(respReadWriter)
require.NoError(t, err)
require.Equal(t, msg, string(receivedMsg))
}
type respReadWriteFlusher struct {
io.Reader
w io.Writer
headers http.Header
statusCode int
setStatusOnce sync.Once
hasStatus chan struct{}
}
func newRespReadWriteFlusher() *respReadWriteFlusher {
pr, pw := io.Pipe()
return &respReadWriteFlusher{
Reader: pr,
w: pw,
headers: make(http.Header),
hasStatus: make(chan struct{}),
}
}
func (rrw *respReadWriteFlusher) Write(buf []byte) (int, error) {
rrw.WriteHeader(http.StatusOK)
return rrw.w.Write(buf)
}
func (rrw *respReadWriteFlusher) Flush() {}
func (rrw *respReadWriteFlusher) Header() http.Header {
return rrw.headers
}
func (rrw *respReadWriteFlusher) WriteHeader(statusCode int) {
rrw.setStatusOnce.Do(func() {
rrw.statusCode = statusCode
close(rrw.hasStatus)
})
}
+1 -10
View File
@@ -1,4 +1,4 @@
package origin
package proxy
import (
"github.com/prometheus/client_golang/prometheus"
@@ -43,14 +43,6 @@ var (
Help: "Count of error proxying to origin",
},
)
haConnections = prometheus.NewGauge(
prometheus.GaugeOpts{
Namespace: connection.MetricsNamespace,
Subsystem: connection.TunnelSubsystem,
Name: "ha_connections",
Help: "Number of active ha connections",
},
)
)
func init() {
@@ -59,7 +51,6 @@ func init() {
concurrentRequests,
responseByCode,
requestErrors,
haConnections,
)
}
+1 -1
View File
@@ -1,4 +1,4 @@
package origin
package proxy
import (
"sync"
+9 -4
View File
@@ -1,4 +1,4 @@
package origin
package proxy
import (
"bufio"
@@ -38,17 +38,22 @@ type Proxy struct {
// NewOriginProxy returns a new instance of the Proxy struct.
func NewOriginProxy(
ingressRules ingress.Ingress,
warpRouting *ingress.WarpRoutingService,
warpRoutingEnabled bool,
tags []tunnelpogs.Tag,
log *zerolog.Logger,
) *Proxy {
return &Proxy{
proxy := &Proxy{
ingressRules: ingressRules,
warpRouting: warpRouting,
tags: tags,
log: log,
bufferPool: newBufferPool(512 * 1024),
}
if warpRoutingEnabled {
proxy.warpRouting = ingress.NewWarpRoutingService()
log.Info().Msgf("Warp-routing is enabled")
}
return proxy
}
// ProxyHTTP further depends on ingress rules to establish a connection with the origin service. This may be
@@ -1,7 +1,7 @@
//go:build !windows
// +build !windows
package origin
package proxy
import (
"io/ioutil"
+10 -18
View File
@@ -1,4 +1,4 @@
package origin
package proxy
import (
"bytes"
@@ -31,8 +31,7 @@ import (
)
var (
testTags = []tunnelpogs.Tag{tunnelpogs.Tag{Name: "Name", Value: "value"}}
unusedWarpRoutingService = (*ingress.WarpRoutingService)(nil)
testTags = []tunnelpogs.Tag{tunnelpogs.Tag{Name: "Name", Value: "value"}}
)
type mockHTTPRespWriter struct {
@@ -131,17 +130,14 @@ func TestProxySingleOrigin(t *testing.T) {
ingressRule, err := ingress.NewSingleOrigin(cliCtx, allowURLFromArgs)
require.NoError(t, err)
var wg sync.WaitGroup
errC := make(chan error)
require.NoError(t, ingressRule.StartOrigins(&wg, &log, ctx.Done(), errC))
require.NoError(t, ingressRule.StartOrigins(&log, ctx.Done()))
proxy := NewOriginProxy(ingressRule, unusedWarpRoutingService, testTags, &log)
proxy := NewOriginProxy(ingressRule, false, testTags, &log)
t.Run("testProxyHTTP", testProxyHTTP(proxy))
t.Run("testProxyWebsocket", testProxyWebsocket(proxy))
t.Run("testProxySSE", testProxySSE(proxy))
t.Run("testProxySSEAllData", testProxySSEAllData(proxy))
cancel()
wg.Wait()
}
func testProxyHTTP(proxy connection.OriginProxy) func(t *testing.T) {
@@ -341,11 +337,9 @@ func runIngressTestScenarios(t *testing.T, unvalidatedIngress []config.Unvalidat
log := zerolog.Nop()
ctx, cancel := context.WithCancel(context.Background())
errC := make(chan error)
var wg sync.WaitGroup
require.NoError(t, ingress.StartOrigins(&wg, &log, ctx.Done(), errC))
require.NoError(t, ingress.StartOrigins(&log, ctx.Done()))
proxy := NewOriginProxy(ingress, unusedWarpRoutingService, testTags, &log)
proxy := NewOriginProxy(ingress, false, testTags, &log)
for _, test := range tests {
responseWriter := newMockHTTPRespWriter()
@@ -363,7 +357,6 @@ func runIngressTestScenarios(t *testing.T, unvalidatedIngress []config.Unvalidat
}
}
cancel()
wg.Wait()
}
type mockAPI struct{}
@@ -394,7 +387,7 @@ func TestProxyError(t *testing.T) {
log := zerolog.Nop()
proxy := NewOriginProxy(ing, unusedWarpRoutingService, testTags, &log)
proxy := NewOriginProxy(ing, false, testTags, &log)
responseWriter := newMockHTTPRespWriter()
req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1", nil)
@@ -634,10 +627,9 @@ func TestConnections(t *testing.T) {
test.args.originService(t, ln)
ingressRule := createSingleIngressConfig(t, test.args.ingressServiceScheme+ln.Addr().String())
var wg sync.WaitGroup
errC := make(chan error)
ingressRule.StartOrigins(&wg, logger, ctx.Done(), errC)
proxy := NewOriginProxy(ingressRule, test.args.warpRoutingService, testTags, logger)
ingressRule.StartOrigins(logger, ctx.Done())
proxy := NewOriginProxy(ingressRule, true, testTags, logger)
proxy.warpRouting = test.args.warpRoutingService
dest := ln.Addr().String()
req, err := http.NewRequest(
+8 -4
View File
@@ -231,14 +231,18 @@ type mockConfigRPCServer struct {
config []byte
}
func (s mockConfigRPCServer) UpdateConfiguration(_ context.Context, version int32, config []byte) (*tunnelpogs.UpdateConfigurationResponse, error) {
func (s mockConfigRPCServer) UpdateConfiguration(_ context.Context, version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
if s.version != version {
return nil, fmt.Errorf("expect version %d, got %d", s.version, version)
return &tunnelpogs.UpdateConfigurationResponse{
Err: fmt.Errorf("expect version %d, got %d", s.version, version),
}
}
if !bytes.Equal(s.config, config) {
return nil, fmt.Errorf("expect config %v, got %v", s.config, config)
return &tunnelpogs.UpdateConfigurationResponse{
Err: fmt.Errorf("expect config %v, got %v", s.config, config),
}
}
return &tunnelpogs.UpdateConfigurationResponse{LastAppliedVersion: version}, nil
return &tunnelpogs.UpdateConfigurationResponse{LastAppliedVersion: version}
}
type mockRPCStream struct {
@@ -1,4 +1,4 @@
package origin
package supervisor
import (
"encoding/json"
@@ -1,4 +1,4 @@
package origin
package supervisor
import (
"testing"
@@ -1,4 +1,4 @@
package origin
package supervisor
import (
"github.com/rs/zerolog"
@@ -1,4 +1,4 @@
package origin
package supervisor
import (
"time"
+27
View File
@@ -0,0 +1,27 @@
package supervisor
import (
"github.com/prometheus/client_golang/prometheus"
"github.com/cloudflare/cloudflared/connection"
)
// Metrics uses connection.MetricsNamespace(aka cloudflared) as namespace and connection.TunnelSubsystem
// (tunnel) as subsystem to keep them consistent with the previous qualifier.
var (
haConnections = prometheus.NewGauge(
prometheus.GaugeOpts{
Namespace: connection.MetricsNamespace,
Subsystem: connection.TunnelSubsystem,
Name: "ha_connections",
Help: "Number of active ha connections",
},
)
)
func init() {
prometheus.MustRegister(
haConnections,
)
}
@@ -1,4 +1,4 @@
package origin
package supervisor
import (
"context"
@@ -1,4 +1,4 @@
package origin
package supervisor
import (
"context"
@@ -1,4 +1,4 @@
package origin
package supervisor
import (
"context"
@@ -13,6 +13,7 @@ import (
"github.com/cloudflare/cloudflared/edgediscovery"
"github.com/cloudflare/cloudflared/edgediscovery/allregions"
"github.com/cloudflare/cloudflared/h2mux"
"github.com/cloudflare/cloudflared/orchestration"
"github.com/cloudflare/cloudflared/retry"
"github.com/cloudflare/cloudflared/signal"
tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
@@ -38,6 +39,7 @@ const (
type Supervisor struct {
cloudflaredUUID uuid.UUID
config *TunnelConfig
orchestrator *orchestration.Orchestrator
edgeIPs *edgediscovery.Edge
tunnelErrors chan tunnelError
tunnelsConnecting map[int]chan struct{}
@@ -64,7 +66,7 @@ type tunnelError struct {
err error
}
func NewSupervisor(config *TunnelConfig, reconnectCh chan ReconnectSignal, gracefulShutdownC <-chan struct{}) (*Supervisor, error) {
func NewSupervisor(config *TunnelConfig, orchestrator *orchestration.Orchestrator, reconnectCh chan ReconnectSignal, gracefulShutdownC <-chan struct{}) (*Supervisor, error) {
cloudflaredUUID, err := uuid.NewRandom()
if err != nil {
return nil, fmt.Errorf("failed to generate cloudflared instance ID: %w", err)
@@ -88,6 +90,7 @@ func NewSupervisor(config *TunnelConfig, reconnectCh chan ReconnectSignal, grace
return &Supervisor{
cloudflaredUUID: cloudflaredUUID,
config: config,
orchestrator: orchestrator,
edgeIPs: edgeIPs,
tunnelErrors: make(chan tunnelError),
tunnelsConnecting: map[int]chan struct{}{},
@@ -243,6 +246,7 @@ func (s *Supervisor) startFirstTunnel(
ctx,
s.reconnectCredentialManager,
s.config,
s.orchestrator,
addr,
s.log,
firstConnIndex,
@@ -277,6 +281,7 @@ func (s *Supervisor) startFirstTunnel(
ctx,
s.reconnectCredentialManager,
s.config,
s.orchestrator,
addr,
s.log,
firstConnIndex,
@@ -311,6 +316,7 @@ func (s *Supervisor) startTunnel(
ctx,
s.reconnectCredentialManager,
s.config,
s.orchestrator,
addr,
s.log,
uint8(index),
@@ -380,7 +386,7 @@ func (s *Supervisor) authenticate(ctx context.Context, numPreviousAttempts int)
defer rpcClient.Close()
const arbitraryConnectionID = uint8(0)
registrationOptions := s.config.RegistrationOptions(arbitraryConnectionID, edgeConn.LocalAddr().String(), s.cloudflaredUUID)
registrationOptions := s.config.registrationOptions(arbitraryConnectionID, edgeConn.LocalAddr().String(), s.cloudflaredUUID)
registrationOptions.NumPreviousAttempts = uint8(numPreviousAttempts)
return rpcClient.Authenticate(ctx, s.config.ClassicTunnel, registrationOptions)
}
+47 -33
View File
@@ -1,4 +1,4 @@
package origin
package supervisor
import (
"context"
@@ -20,6 +20,7 @@ import (
"github.com/cloudflare/cloudflared/edgediscovery"
"github.com/cloudflare/cloudflared/edgediscovery/allregions"
"github.com/cloudflare/cloudflared/h2mux"
"github.com/cloudflare/cloudflared/orchestration"
quicpogs "github.com/cloudflare/cloudflared/quic"
"github.com/cloudflare/cloudflared/retry"
"github.com/cloudflare/cloudflared/signal"
@@ -34,32 +35,33 @@ const (
)
type TunnelConfig struct {
ConnectionConfig *connection.Config
OSArch string
ClientID string
CloseConnOnce *sync.Once // Used to close connectedSignal no more than once
EdgeAddrs []string
Region string
HAConnections int
IncidentLookup IncidentLookup
IsAutoupdated bool
LBPool string
Tags []tunnelpogs.Tag
Log *zerolog.Logger
LogTransport *zerolog.Logger
Observer *connection.Observer
ReportedVersion string
Retries uint
RunFromTerminal bool
GracePeriod time.Duration
ReplaceExisting bool
OSArch string
ClientID string
CloseConnOnce *sync.Once // Used to close connectedSignal no more than once
EdgeAddrs []string
Region string
HAConnections int
IncidentLookup IncidentLookup
IsAutoupdated bool
LBPool string
Tags []tunnelpogs.Tag
Log *zerolog.Logger
LogTransport *zerolog.Logger
Observer *connection.Observer
ReportedVersion string
Retries uint
RunFromTerminal bool
NamedTunnel *connection.NamedTunnelConfig
ClassicTunnel *connection.ClassicTunnelConfig
NamedTunnel *connection.NamedTunnelProperties
ClassicTunnel *connection.ClassicTunnelProperties
MuxerConfig *connection.MuxerConfig
ProtocolSelector connection.ProtocolSelector
EdgeTLSConfigs map[connection.Protocol]*tls.Config
}
func (c *TunnelConfig) RegistrationOptions(connectionID uint8, OriginLocalIP string, uuid uuid.UUID) *tunnelpogs.RegistrationOptions {
func (c *TunnelConfig) registrationOptions(connectionID uint8, OriginLocalIP string, uuid uuid.UUID) *tunnelpogs.RegistrationOptions {
policy := tunnelrpc.ExistingTunnelPolicy_balance
if c.HAConnections <= 1 && c.LBPool == "" {
policy = tunnelrpc.ExistingTunnelPolicy_disconnect
@@ -81,7 +83,7 @@ func (c *TunnelConfig) RegistrationOptions(connectionID uint8, OriginLocalIP str
}
}
func (c *TunnelConfig) ConnectionOptions(originLocalAddr string, numPreviousAttempts uint8) *tunnelpogs.ConnectionOptions {
func (c *TunnelConfig) connectionOptions(originLocalAddr string, numPreviousAttempts uint8) *tunnelpogs.ConnectionOptions {
// attempt to parse out origin IP, but don't fail since it's informational field
host, _, _ := net.SplitHostPort(originLocalAddr)
originIP := net.ParseIP(host)
@@ -89,7 +91,7 @@ func (c *TunnelConfig) ConnectionOptions(originLocalAddr string, numPreviousAtte
return &tunnelpogs.ConnectionOptions{
Client: c.NamedTunnel.Client,
OriginLocalIP: originIP,
ReplaceExisting: c.ConnectionConfig.ReplaceExisting,
ReplaceExisting: c.ReplaceExisting,
CompressionQuality: uint8(c.MuxerConfig.CompressionSetting),
NumPreviousAttempts: numPreviousAttempts,
}
@@ -106,11 +108,12 @@ func (c *TunnelConfig) SupportedFeatures() []string {
func StartTunnelDaemon(
ctx context.Context,
config *TunnelConfig,
orchestrator *orchestration.Orchestrator,
connectedSignal *signal.Signal,
reconnectCh chan ReconnectSignal,
graceShutdownC <-chan struct{},
) error {
s, err := NewSupervisor(config, reconnectCh, graceShutdownC)
s, err := NewSupervisor(config, orchestrator, reconnectCh, graceShutdownC)
if err != nil {
return err
}
@@ -121,6 +124,7 @@ func ServeTunnelLoop(
ctx context.Context,
credentialManager *reconnectCredentialManager,
config *TunnelConfig,
orchestrator *orchestration.Orchestrator,
addr *allregions.EdgeAddr,
connAwareLogger *ConnAwareLogger,
connIndex uint8,
@@ -156,6 +160,7 @@ func ServeTunnelLoop(
connLog,
credentialManager,
config,
orchestrator,
addr,
connIndex,
connectedFuse,
@@ -254,6 +259,7 @@ func ServeTunnel(
connLog *ConnAwareLogger,
credentialManager *reconnectCredentialManager,
config *TunnelConfig,
orchestrator *orchestration.Orchestrator,
addr *allregions.EdgeAddr,
connIndex uint8,
fuse *h2mux.BooleanFuse,
@@ -282,6 +288,7 @@ func ServeTunnel(
connLog,
credentialManager,
config,
orchestrator,
addr,
connIndex,
fuse,
@@ -330,6 +337,7 @@ func serveTunnel(
connLog *ConnAwareLogger,
credentialManager *reconnectCredentialManager,
config *TunnelConfig,
orchestrator *orchestration.Orchestrator,
addr *allregions.EdgeAddr,
connIndex uint8,
fuse *h2mux.BooleanFuse,
@@ -339,7 +347,6 @@ func serveTunnel(
protocol connection.Protocol,
gracefulShutdownC <-chan struct{},
) (err error, recoverable bool) {
connectedFuse := &connectedFuse{
fuse: fuse,
backoff: backoff,
@@ -351,15 +358,16 @@ func serveTunnel(
connIndex,
nil,
gracefulShutdownC,
config.ConnectionConfig.GracePeriod,
config.GracePeriod,
)
switch protocol {
case connection.QUIC, connection.QUICWarp:
connOptions := config.ConnectionOptions(addr.UDP.String(), uint8(backoff.Retries()))
connOptions := config.connectionOptions(addr.UDP.String(), uint8(backoff.Retries()))
return ServeQUIC(ctx,
addr.UDP,
config,
orchestrator,
connLog,
connOptions,
controlStream,
@@ -374,11 +382,12 @@ func serveTunnel(
return err, true
}
connOptions := config.ConnectionOptions(edgeConn.LocalAddr().String(), uint8(backoff.Retries()))
connOptions := config.connectionOptions(edgeConn.LocalAddr().String(), uint8(backoff.Retries()))
if err := ServeHTTP2(
ctx,
connLog,
config,
orchestrator,
edgeConn,
connOptions,
controlStream,
@@ -401,6 +410,7 @@ func serveTunnel(
connLog,
credentialManager,
config,
orchestrator,
edgeConn,
connIndex,
connectedFuse,
@@ -427,6 +437,7 @@ func ServeH2mux(
connLog *ConnAwareLogger,
credentialManager *reconnectCredentialManager,
config *TunnelConfig,
orchestrator *orchestration.Orchestrator,
edgeConn net.Conn,
connIndex uint8,
connectedFuse *connectedFuse,
@@ -437,7 +448,8 @@ func ServeH2mux(
connLog.Logger().Debug().Msgf("Connecting via h2mux")
// Returns error from parsing the origin URL or handshake errors
handler, err, recoverable := connection.NewH2muxConnection(
config.ConnectionConfig,
orchestrator,
config.GracePeriod,
config.MuxerConfig,
edgeConn,
connIndex,
@@ -455,10 +467,10 @@ func ServeH2mux(
errGroup.Go(func() error {
if config.NamedTunnel != nil {
connOptions := config.ConnectionOptions(edgeConn.LocalAddr().String(), uint8(connectedFuse.backoff.Retries()))
connOptions := config.connectionOptions(edgeConn.LocalAddr().String(), uint8(connectedFuse.backoff.Retries()))
return handler.ServeNamedTunnel(serveCtx, config.NamedTunnel, connOptions, connectedFuse)
}
registrationOptions := config.RegistrationOptions(connIndex, edgeConn.LocalAddr().String(), cloudflaredUUID)
registrationOptions := config.registrationOptions(connIndex, edgeConn.LocalAddr().String(), cloudflaredUUID)
return handler.ServeClassicTunnel(serveCtx, config.ClassicTunnel, credentialManager, registrationOptions, connectedFuse)
})
@@ -473,6 +485,7 @@ func ServeHTTP2(
ctx context.Context,
connLog *ConnAwareLogger,
config *TunnelConfig,
orchestrator *orchestration.Orchestrator,
tlsServerConn net.Conn,
connOptions *tunnelpogs.ConnectionOptions,
controlStreamHandler connection.ControlStreamHandler,
@@ -483,7 +496,7 @@ func ServeHTTP2(
connLog.Logger().Debug().Msgf("Connecting via http2")
h2conn := connection.NewHTTP2Connection(
tlsServerConn,
config.ConnectionConfig,
orchestrator,
connOptions,
config.Observer,
connIndex,
@@ -512,6 +525,7 @@ func ServeQUIC(
ctx context.Context,
edgeAddr *net.UDPAddr,
config *TunnelConfig,
orchestrator *orchestration.Orchestrator,
connLogger *ConnAwareLogger,
connOptions *tunnelpogs.ConnectionOptions,
controlStreamHandler connection.ControlStreamHandler,
@@ -535,7 +549,7 @@ func ServeQUIC(
quicConfig,
edgeAddr,
tlsConfig,
config.ConnectionConfig.OriginProxy,
orchestrator,
connOptions,
controlStreamHandler,
connLogger.Logger())
@@ -1,4 +1,4 @@
package origin
package supervisor
import (
"testing"
@@ -32,11 +32,7 @@ func TestWaitForBackoffFallback(t *testing.T) {
}
log := zerolog.Nop()
resolveTTL := time.Duration(0)
namedTunnel := &connection.NamedTunnelConfig{
Credentials: connection.Credentials{
AccountTag: "test-account",
},
}
namedTunnel := &connection.NamedTunnelProperties{}
mockFetcher := dynamicMockFetcher{
protocolPercents: edgediscovery.ProtocolPercents{edgediscovery.ProtocolPercent{Protocol: "http2", Percentage: 100}},
}
@@ -1,4 +1,4 @@
package origin
package supervisor
import (
"fmt"
+2 -6
View File
@@ -11,7 +11,7 @@ import (
)
type ConfigurationManager interface {
UpdateConfiguration(ctx context.Context, version int32, config []byte) (*UpdateConfigurationResponse, error)
UpdateConfiguration(ctx context.Context, version int32, config []byte) *UpdateConfigurationResponse
}
type ConfigurationManager_PogsImpl struct {
@@ -31,16 +31,12 @@ func (i ConfigurationManager_PogsImpl) UpdateConfiguration(p tunnelrpc.Configura
return err
}
updateResp, err := i.impl.UpdateConfiguration(p.Ctx, version, config)
if err != nil {
return err
}
result, err := p.Results.NewResult()
if err != nil {
return err
}
updateResp := i.impl.UpdateConfiguration(p.Ctx, version, config)
return updateResp.Marshal(result)
}
+38 -30
View File
@@ -3,8 +3,10 @@ package websocket
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"sync"
"time"
gobwas "github.com/gobwas/ws"
@@ -14,9 +16,6 @@ import (
)
const (
// Time allowed to write a message to the peer.
writeWait = 10 * time.Second
// Time allowed to read the next pong message from the peer.
defaultPongWait = 60 * time.Second
@@ -79,34 +78,20 @@ func (c *GorillaConn) SetDeadline(t time.Time) error {
return nil
}
// pinger simulates the websocket connection to keep it alive
func (c *GorillaConn) pinger(ctx context.Context) {
ticker := time.NewTicker(defaultPingPeriod)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if err := c.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(writeWait)); err != nil {
c.log.Debug().Msgf("failed to send ping message: %s", err)
}
case <-ctx.Done():
return
}
}
}
type Conn struct {
rw io.ReadWriter
log *zerolog.Logger
// closed is a channel to indicate if Conn has been fully terminated
shutdownC chan struct{}
// writeLock makes sure
// 1. Only one write at a time. The pinger and Stream function can both call write.
// 2. Close only returns after in progress Write is finished, and no more Write will succeed after calling Close.
writeLock sync.Mutex
done bool
}
func NewConn(ctx context.Context, rw io.ReadWriter, log *zerolog.Logger) *Conn {
c := &Conn{
rw: rw,
log: log,
shutdownC: make(chan struct{}),
rw: rw,
log: log,
}
go c.pinger(ctx)
return c
@@ -121,16 +106,22 @@ func (c *Conn) Read(reader []byte) (int, error) {
return copy(reader, data), nil
}
// Write will write messages to the websocket connection
// Write will write messages to the websocket connection.
// It will not write to the connection after Close is called to fix TUN-5184
func (c *Conn) Write(p []byte) (int, error) {
c.writeLock.Lock()
defer c.writeLock.Unlock()
if c.done {
return 0, errors.New("write to closed websocket connection")
}
if err := wsutil.WriteServerBinary(c.rw, p); err != nil {
return 0, err
}
return len(p), nil
}
func (c *Conn) pinger(ctx context.Context) {
defer close(c.shutdownC)
pongMessge := wsutil.Message{
OpCode: gobwas.OpPong,
Payload: []byte{},
@@ -141,7 +132,11 @@ func (c *Conn) pinger(ctx context.Context) {
for {
select {
case <-ticker.C:
if err := wsutil.WriteServerMessage(c.rw, gobwas.OpPing, []byte{}); err != nil {
done, err := c.ping()
if done {
return
}
if err != nil {
c.log.Debug().Err(err).Msgf("failed to write ping message")
}
if err := wsutil.HandleClientControlMessage(c.rw, pongMessge); err != nil {
@@ -153,6 +148,17 @@ func (c *Conn) pinger(ctx context.Context) {
}
}
func (c *Conn) ping() (bool, error) {
c.writeLock.Lock()
defer c.writeLock.Unlock()
if c.done {
return true, nil
}
return false, wsutil.WriteServerMessage(c.rw, gobwas.OpPing, []byte{})
}
func (c *Conn) pingPeriod(ctx context.Context) time.Duration {
if val := ctx.Value(PingPeriodContextKey); val != nil {
if period, ok := val.(time.Duration); ok {
@@ -162,7 +168,9 @@ func (c *Conn) pingPeriod(ctx context.Context) time.Duration {
return defaultPingPeriod
}
// Close waits for pinger to terminate
func (c *Conn) WaitForShutdown() {
<-c.shutdownC
// Close waits for the current write to finish. Further writes will return error
func (c *Conn) Close() {
c.writeLock.Lock()
defer c.writeLock.Unlock()
c.done = true
}
+54 -17
View File
@@ -8,6 +8,7 @@ import (
"fmt"
"io"
"net/http"
"sync/atomic"
"time"
"github.com/gorilla/websocket"
@@ -28,28 +29,64 @@ func NewResponseHeader(req *http.Request) http.Header {
return header
}
type bidirectionalStreamStatus struct {
doneChan chan struct{}
anyDone uint32
}
func newBiStreamStatus() *bidirectionalStreamStatus {
return &bidirectionalStreamStatus{
doneChan: make(chan struct{}, 2),
anyDone: 0,
}
}
func (s *bidirectionalStreamStatus) markUniStreamDone() {
atomic.StoreUint32(&s.anyDone, 1)
s.doneChan <- struct{}{}
}
func (s *bidirectionalStreamStatus) waitAnyDone() {
<-s.doneChan
}
func (s *bidirectionalStreamStatus) isAnyDone() bool {
return atomic.LoadUint32(&s.anyDone) > 0
}
// Stream copies copy data to & from provided io.ReadWriters.
func Stream(tunnelConn, originConn io.ReadWriter, log *zerolog.Logger) {
proxyDone := make(chan struct{}, 2)
status := newBiStreamStatus()
go func() {
_, err := copyData(tunnelConn, originConn, "origin->tunnel")
if err != nil {
log.Debug().Msgf("origin to tunnel copy: %v", err)
}
proxyDone <- struct{}{}
}()
go func() {
_, err := copyData(originConn, tunnelConn, "tunnel->origin")
if err != nil {
log.Debug().Msgf("tunnel to origin copy: %v", err)
}
proxyDone <- struct{}{}
}()
go unidirectionalStream(tunnelConn, originConn, "origin->tunnel", status, log)
go unidirectionalStream(originConn, tunnelConn, "tunnel->origin", status, log)
// If one side is done, we are done.
<-proxyDone
status.waitAnyDone()
}
func unidirectionalStream(dst io.Writer, src io.Reader, dir string, status *bidirectionalStreamStatus, log *zerolog.Logger) {
defer func() {
// The bidirectional streaming spawns 2 goroutines to stream each direction.
// If any ends, the callstack returns, meaning the Tunnel request/stream (depending on http2 vs quic) will
// close. In such case, if the other direction did not stop (due to application level stopping, e.g., if a
// server/origin listens forever until closure), it may read/write from the underlying ReadWriter (backed by
// the Edge<->cloudflared transport) in an unexpected state.
if status.isAnyDone() {
// Because of this, we set this recover() logic, which kicks-in *only* if any stream is known to have
// exited. In such case, we stop a possible panic from propagating upstream.
if r := recover(); r != nil {
// We handle such unexpected errors only when we detect that one side of the streaming is done.
log.Debug().Msgf("Handled gracefully error %v in Streaming for %s", r, dir)
}
}
}()
_, err := copyData(dst, src, dir)
if err != nil {
log.Debug().Msgf("%s copy: %v", dir, err)
}
status.markUniStreamDone()
}
// when set to true, enables logging of content copied to/from origin and tunnel