mirror of
https://github.com/openziti/ziti.git
synced 2026-09-10 00:35:41 +00:00
Paul Lorenz
67eba590e2
Implements the router-side Connect-V2 sessionless dial path. Dials are authorized locally via the RouterDataModel instead of a controller-issued service session token; circuit creation flows through the existing `CreateCircuitV3` controller endpoint (#3721). Builds on the sdk-golang v2 migration. - Adds `processConnectV2` on `edgeClientConn`: resolves the service by id or name via the RouterDataModel, checks dial access, and dispatches to the controller via `sendCreateCircuitV3Msg`. Supports both `xgEdgeForwarder` (SDK xgress) and `nonXgConnectHandler` flow-control modes, selected by the SDK's `UseXgressToSdkHeader`. - Makes `CircuitId` optional in `DecodeCreateCircuitV3Request`. The V2 router path does not pre-assign a circuit ID; the controller generates it as V1/V2 already do. Without this the decoder rejected the empty header and every V2 dial hung until timeout. Adds a regression test. - Splits `checkAccess` to close a posture-check bypass on the V2 path. The old single `checkAccess` short-circuited to nil for non-OIDC sessions (V1 ran posture at the controller during `CreateSession`); V2 has no such step, so posture would have been skipped. `checkAccess` now always runs the RDM `HasAccess` (policy + posture) check; `checkAccessIfOidc` keeps the OIDC-only gate for the V1 and bind paths. - Sends the V2 `state_connected` on the default (data) sender rather than the control sender. On multi-underlay channels the two senders are independently ordered, so an early terminator payload on the data sender could beat `state_connected` to the SDK and be dropped (channel/v5 has no message-priority API). - Updates `xgEdgeForwarder.lastRx` on every forward path, including the fast `timeout == 0` `TrySend` branch used for normal payload dispatch. The old code only updated it on the `timeout > 0` path, so active V2 circuits looked idle and could be unrouted prematurely. - Adds `state.ConnState.ServiceId`, populated by the connect handlers from the service session token (V1) or the request header (V2). The non-xgress V2 path previously left this empty, so `handleDialAccessLost` could not identify and close V2 non-xgress circuits when dial access was revoked. - Skips conns with no `ServiceSessionToken` in `RemoveLegacyServiceSession`; a sessionless V2 conn's token is nil and the cleanup loop previously dereferenced it unconditionally, which would panic the router. - Advertises Connect-V2 via the `RouterCapabilityConnectV2` bit in the listener hello so SDKs can detect V2 support. - Wires `ContentTypeConnectV2` and `ContentTypeXgControl` handlers in `Acceptor.BindChannel`, and adds `handleXgControl` for SDK-side xgress control messages, preserving `ControlUserVal` so trace-route responses correlate back to the initiator's `SendForReply` waiter. - Adds `RouterDataModel.serviceNameIndex` for O(1) name->id lookup in the V2 dial path, maintained with rename safety at the `HandleServiceEvent` mutation points. - Adds `tests/connect_v2_test.go` covering end-to-end V2 dataflow and the V1 fallback (`ForceConnectV1`), asserting the dial path via the SDK `DialEvent`. - Propagates a V2 initiator's graceful half-close to legacy hosts via `edgeXgressConn.FlowFromFabricToXgressClosed`, which emits an edge FIN when the fabric->app half of the circuit closes. The SDK signals half-close to its router xgress peer with the native xgress EOF flag; without translating that to an edge FIN, a legacy host reading to EOF stalled until teardown. - Records the dialing identity id as the circuit `ClientId` for sessionless V2 dials, since there is no dial session to key on; updates `Test_OidcEvents` to match. - Adds `tests/connect_v2_teardown_test.go` covering client- and host-initiated close propagation on both the V2 and forced-V1 paths. - Polls for the asynchronous conn close in the SDK posture-check tests (`awaitClientConnClosed`): revocation tears the circuit down out of band, so checking `IsClosed` immediately after the first read error was racy. - Temporarily pins sdk-golang/v2 to the openziti/sdk-golang#959 commit, which carries the matching xgress conn-close-on-teardown fix the V2 posture tests depend on; to be repointed at the next sdk-golang pre-release before merge. For openziti/sdk-golang#936.
390 lines
14 KiB
Go
390 lines
14 KiB
Go
//go:build apitests
|
|
|
|
/*
|
|
Copyright NetFoundry Inc.
|
|
|
|
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
|
|
|
|
https://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.
|
|
*/
|
|
|
|
package tests
|
|
|
|
import (
|
|
"fmt"
|
|
"reflect"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/openziti/foundation/v2/stringz"
|
|
"github.com/openziti/sdk-golang/v2/ziti"
|
|
"github.com/openziti/ziti/v2/controller/event"
|
|
"github.com/openziti/ziti/v2/controller/events"
|
|
"github.com/openziti/ziti/v2/controller/xt_smartrouting"
|
|
"github.com/openziti/ziti/v2/router/env"
|
|
)
|
|
|
|
type eventsCollector struct {
|
|
sync.Mutex
|
|
events chan interface{}
|
|
}
|
|
|
|
func (self *eventsCollector) acceptEvent(event interface{}) {
|
|
self.events <- event
|
|
fmt.Printf("\nNEXT EVENT: %v: %v %+v\n", reflect.TypeOf(event), event, event)
|
|
}
|
|
|
|
func (self *eventsCollector) AcceptUsageEvent(event *event.UsageEventV2) {
|
|
self.acceptEvent(event)
|
|
}
|
|
|
|
func (self *eventsCollector) AcceptSessionEvent(event *event.SessionEvent) {
|
|
self.acceptEvent(event)
|
|
}
|
|
|
|
func (self *eventsCollector) AcceptCircuitEvent(event *event.CircuitEvent) {
|
|
self.acceptEvent(event)
|
|
}
|
|
|
|
func (self *eventsCollector) AcceptApiSessionEvent(event *event.ApiSessionEvent) {
|
|
self.acceptEvent(event)
|
|
}
|
|
|
|
func (self *eventsCollector) PopNextEvent(ctx *TestContext, desc string, timeout time.Duration) interface{} {
|
|
select {
|
|
case evt := <-self.events:
|
|
return evt
|
|
case <-time.After(timeout):
|
|
ctx.Fail("timed out waiting for event", desc)
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func Test_LegacyEvents(t *testing.T) {
|
|
ctx := NewTestContext(t)
|
|
defer ctx.Teardown()
|
|
|
|
ctx.StartServer()
|
|
|
|
ctx.RequireAdminManagementApiLogin()
|
|
ctx.RequireAdminClientApiLogin()
|
|
|
|
ec := &eventsCollector{
|
|
events: make(chan interface{}, 50),
|
|
}
|
|
|
|
dispatcher := ctx.fabricController.GetEventDispatcher()
|
|
|
|
dispatcher.AddApiSessionEventHandler(ec)
|
|
defer dispatcher.RemoveApiSessionEventHandler(ec)
|
|
|
|
dispatcher.AddCircuitEventHandler(ec)
|
|
defer dispatcher.RemoveCircuitEventHandler(ec)
|
|
|
|
dispatcher.AddSessionEventHandler(ec)
|
|
defer dispatcher.RemoveSessionEventHandler(ec)
|
|
|
|
dispatcher.AddUsageEventHandler(ec)
|
|
defer dispatcher.RemoveUsageEventHandler(ec)
|
|
|
|
ctx.CreateEnrollAndStartEdgeRouterWithCfgTweaks(func(config *env.Config) {
|
|
config.Metrics.ReportInterval = time.Second * 5
|
|
config.Metrics.IntervalAgeThreshold = time.Second * 6
|
|
})
|
|
|
|
service := ctx.AdminManagementSession.RequireNewServiceAccessibleToAll(xt_smartrouting.Name)
|
|
|
|
hostIdentity, hostContext := ctx.AdminManagementSession.RequireCreateSdkContext()
|
|
defer hostContext.Close()
|
|
|
|
// We're testing legacy/non-oidc authentication, so we need to disable OIDC
|
|
hostContext.(*ziti.ContextImpl).CtrlClt.SetAllowOidcDynamicallyEnabled(false)
|
|
|
|
listener, err := hostContext.Listen(service.Name)
|
|
ctx.Req.NoError(err)
|
|
defer func() { _ = listener.Close() }()
|
|
|
|
testServer := newTestServer(listener, func(conn *testServerConn) error {
|
|
conn.ReadString(128, time.Second)
|
|
return conn.server.close()
|
|
})
|
|
testServer.start()
|
|
|
|
clientIdentity, clientContext := ctx.AdminManagementSession.RequireCreateSdkContext()
|
|
defer clientContext.Close()
|
|
|
|
// We're testing legacy/non-oidc authentication, so we need to disable OIDC
|
|
clientContext.(*ziti.ContextImpl).CtrlClt.SetAllowOidcDynamicallyEnabled(false)
|
|
|
|
conn := ctx.WrapConn(clientContext.Dial(service.Name))
|
|
defer func() { _ = conn.Close() }()
|
|
|
|
conn.WriteString("hello, hello, how are you?", time.Second)
|
|
|
|
testServer.waitForDone(ctx, 5*time.Second)
|
|
// TODO: Figure out how to make this test faster. Was using ctx.router.GetMetricsRegistry().Flush(), but it's not ideal
|
|
ctx.Req.NoError(err)
|
|
|
|
evt := ec.PopNextEvent(ctx, "api.sessions.created", time.Second)
|
|
apiSession, ok := evt.(*event.ApiSessionEvent)
|
|
ctx.Req.Truef(ok, "should have been api session event, instead of %T", evt)
|
|
ctx.Req.Equal("apiSession", apiSession.Namespace)
|
|
ctx.Req.Equal("created", apiSession.EventType)
|
|
ctx.Req.Equalf(hostIdentity.Id, apiSession.IdentityId, "host id %s, client id %s", hostIdentity.Id, clientIdentity.Id)
|
|
ctx.Req.Equal("legacy", apiSession.Type)
|
|
|
|
evt = ec.PopNextEvent(ctx, "sessions.created", time.Second)
|
|
edgeSession, ok := evt.(*event.SessionEvent)
|
|
ctx.Req.Truef(ok, "should have been session event, instead of %T", evt)
|
|
ctx.Req.Equal("session", edgeSession.Namespace)
|
|
ctx.Req.Equal("created", edgeSession.EventType)
|
|
ctx.Req.Equal("legacy", edgeSession.Provider)
|
|
ctx.Req.Equal(hostIdentity.Id, edgeSession.IdentityId)
|
|
|
|
evt = ec.PopNextEvent(ctx, "api.sessions.created", time.Second)
|
|
apiSession, ok = evt.(*event.ApiSessionEvent)
|
|
ctx.Req.Truef(ok, "should have been api session event, instead of %T", evt)
|
|
ctx.Req.Equal("apiSession", apiSession.Namespace)
|
|
ctx.Req.Equal("created", apiSession.EventType)
|
|
ctx.Req.Equal("legacy", edgeSession.Provider)
|
|
ctx.Req.Equalf(clientIdentity.Id, apiSession.IdentityId, "host id %s, client id %s", hostIdentity.Id, clientIdentity.Id)
|
|
ctx.Req.Equal("legacy", apiSession.Type)
|
|
|
|
evt = ec.PopNextEvent(ctx, "edge.sessions.created", time.Second)
|
|
edgeSession, ok = evt.(*event.SessionEvent)
|
|
ctx.Req.Truef(ok, "should have been session event, instead of %T", evt)
|
|
ctx.Req.Equal("session", edgeSession.Namespace)
|
|
ctx.Req.Equal("created", edgeSession.EventType)
|
|
ctx.Req.Equal("legacy", edgeSession.Provider)
|
|
ctx.Req.Equal(clientIdentity.Id, edgeSession.IdentityId)
|
|
|
|
evt = ec.PopNextEvent(ctx, "circuits.created", time.Second)
|
|
circuitEvent, ok := evt.(*event.CircuitEvent)
|
|
ctx.Req.True(ok)
|
|
ctx.Req.Equal("circuit", circuitEvent.Namespace)
|
|
ctx.Req.Equal("created", string(circuitEvent.EventType))
|
|
ctx.Req.Equal("legacy", edgeSession.Provider)
|
|
ctx.Req.Equal(service.Id, circuitEvent.ServiceId)
|
|
ctx.Req.Equal(edgeSession.Id, circuitEvent.ClientId)
|
|
|
|
timeout := time.Second * 20
|
|
for i := 0; i < 3; i++ {
|
|
evt = ec.PopNextEvent(ctx, fmt.Sprintf("usage or circuits deleted %v", i+1), timeout)
|
|
if usage, ok := evt.(*event.UsageEventV2); ok {
|
|
ctx.Req.Equal("usage", usage.Namespace)
|
|
ctx.Req.Equal(uint32(2), usage.Version)
|
|
ctx.Req.Equal(circuitEvent.CircuitId, usage.CircuitId)
|
|
expected := []string{"usage.ingress.rx", "usage.egress.tx"}
|
|
ctx.Req.True(stringz.Contains(expected, usage.EventType), "was %v, expected one of %+v", usage.EventType, expected)
|
|
ctx.Req.Equal(ctx.edgeRouterEntity.id, usage.SourceId)
|
|
ctx.Req.Equal(uint64(26), usage.Usage)
|
|
} else if circuitEvent, ok := evt.(*event.CircuitEvent); ok {
|
|
ctx.Req.Equal("circuit", circuitEvent.Namespace)
|
|
ctx.Req.Equal("deleted", string(circuitEvent.EventType))
|
|
ctx.Req.Equal(edgeSession.Id, circuitEvent.ClientId)
|
|
} else {
|
|
ctx.Req.Fail("unexpected event type: %v", reflect.TypeOf(evt))
|
|
}
|
|
}
|
|
}
|
|
|
|
func Test_OidcEvents(t *testing.T) {
|
|
ctx := NewTestContext(t)
|
|
defer ctx.Teardown()
|
|
|
|
ctx.StartServer()
|
|
|
|
ctx.RequireAdminManagementApiLogin()
|
|
ctx.RequireAdminClientApiLogin()
|
|
|
|
ec := &eventsCollector{
|
|
events: make(chan interface{}, 50),
|
|
}
|
|
|
|
dispatcher := ctx.fabricController.GetEventDispatcher()
|
|
|
|
dispatcher.AddApiSessionEventHandler(ec)
|
|
defer dispatcher.RemoveApiSessionEventHandler(ec)
|
|
|
|
dispatcher.AddCircuitEventHandler(ec)
|
|
defer dispatcher.RemoveCircuitEventHandler(ec)
|
|
|
|
dispatcher.AddSessionEventHandler(ec)
|
|
defer dispatcher.RemoveSessionEventHandler(ec)
|
|
|
|
dispatcher.AddUsageEventHandler(ec)
|
|
defer dispatcher.RemoveUsageEventHandler(ec)
|
|
|
|
ctx.CreateEnrollAndStartEdgeRouterWithCfgTweaks(func(config *env.Config) {
|
|
config.Metrics.ReportInterval = time.Second * 5
|
|
config.Metrics.IntervalAgeThreshold = time.Second * 6
|
|
})
|
|
|
|
service := ctx.AdminManagementSession.RequireNewServiceAccessibleToAll(xt_smartrouting.Name)
|
|
|
|
hostIdentity, hostContext := ctx.AdminManagementSession.RequireCreateSdkContext()
|
|
defer hostContext.Close()
|
|
|
|
listener, err := hostContext.Listen(service.Name)
|
|
ctx.Req.NoError(err)
|
|
defer func() { _ = listener.Close() }()
|
|
|
|
testServer := newTestServer(listener, func(conn *testServerConn) error {
|
|
conn.ReadString(128, time.Second)
|
|
return conn.server.close()
|
|
})
|
|
testServer.start()
|
|
|
|
clientIdentity, clientContext := ctx.AdminManagementSession.RequireCreateSdkContext()
|
|
defer clientContext.Close()
|
|
|
|
conn := ctx.WrapConn(clientContext.Dial(service.Name))
|
|
defer func() { _ = conn.Close() }()
|
|
|
|
conn.WriteString("hello, hello, how are you?", time.Second)
|
|
|
|
testServer.waitForDone(ctx, 5*time.Second)
|
|
// TODO: Figure out how to make this test faster. Was using ctx.router.GetMetricsRegistry().Flush(), but it's not ideal
|
|
ctx.Req.NoError(err)
|
|
|
|
evt := ec.PopNextEvent(ctx, "api.sessions.created", time.Second)
|
|
apiSession, ok := evt.(*event.ApiSessionEvent)
|
|
ctx.Req.Truef(ok, "should have been api session event, instead of %T", evt)
|
|
ctx.Req.Equal("apiSession", apiSession.Namespace)
|
|
ctx.Req.Equal("created", apiSession.EventType)
|
|
ctx.Req.Equalf(hostIdentity.Id, apiSession.IdentityId, "host id %s, client id %s", hostIdentity.Id, clientIdentity.Id)
|
|
ctx.Req.Equal("jwt", apiSession.Type)
|
|
|
|
evt = ec.PopNextEvent(ctx, "sessions.created", time.Second)
|
|
edgeSession, ok := evt.(*event.SessionEvent)
|
|
ctx.Req.Truef(ok, "should have been session event, instead of %T", evt)
|
|
ctx.Req.Equal("session", edgeSession.Namespace)
|
|
ctx.Req.Equal("created", edgeSession.EventType)
|
|
ctx.Req.Equal("jwt", edgeSession.Provider)
|
|
ctx.Req.Equal(hostIdentity.Id, edgeSession.IdentityId)
|
|
|
|
evt = ec.PopNextEvent(ctx, "api.sessions.created", time.Second)
|
|
apiSession, ok = evt.(*event.ApiSessionEvent)
|
|
ctx.Req.Truef(ok, "should have been api session event, instead of %T", evt)
|
|
ctx.Req.Equal("apiSession", apiSession.Namespace)
|
|
ctx.Req.Equal("created", apiSession.EventType)
|
|
ctx.Req.Equalf(clientIdentity.Id, apiSession.IdentityId, "host id %s, client id %s", hostIdentity.Id, clientIdentity.Id)
|
|
ctx.Req.Equal("jwt", apiSession.Type)
|
|
|
|
evt = ec.PopNextEvent(ctx, "edge.sessions.created", time.Second)
|
|
edgeSession, ok = evt.(*event.SessionEvent)
|
|
ctx.Req.Truef(ok, "should have been session event, instead of %T", evt)
|
|
ctx.Req.Equal("session", edgeSession.Namespace)
|
|
ctx.Req.Equal("created", edgeSession.EventType)
|
|
ctx.Req.Equal("jwt", edgeSession.Provider)
|
|
ctx.Req.Equal(clientIdentity.Id, edgeSession.IdentityId)
|
|
|
|
evt = ec.PopNextEvent(ctx, "circuits.created", time.Second)
|
|
circuitEvent, ok := evt.(*event.CircuitEvent)
|
|
ctx.Req.True(ok)
|
|
ctx.Req.Equal("circuit", circuitEvent.Namespace)
|
|
ctx.Req.Equal("created", string(circuitEvent.EventType))
|
|
ctx.Req.Equal(service.Id, circuitEvent.ServiceId)
|
|
// ConnectV2 dials are sessionless, so the circuit's ClientId is the dialing
|
|
// identity id rather than an edge (dial) session id.
|
|
ctx.Req.Equal(clientIdentity.Id, circuitEvent.ClientId)
|
|
|
|
timeout := time.Second * 20
|
|
for i := 0; i < 3; i++ {
|
|
evt = ec.PopNextEvent(ctx, fmt.Sprintf("usage or circuits deleted %v", i+1), timeout)
|
|
if usage, ok := evt.(*event.UsageEventV2); ok {
|
|
ctx.Req.Equal("usage", usage.Namespace)
|
|
ctx.Req.Equal(uint32(2), usage.Version)
|
|
ctx.Req.Equal(circuitEvent.CircuitId, usage.CircuitId)
|
|
expected := []string{"usage.ingress.rx", "usage.egress.tx"}
|
|
ctx.Req.True(stringz.Contains(expected, usage.EventType), "was %v, expected one of %+v", usage.EventType, expected)
|
|
ctx.Req.Equal(ctx.edgeRouterEntity.id, usage.SourceId)
|
|
ctx.Req.Equal(uint64(26), usage.Usage)
|
|
} else if circuitEvent, ok := evt.(*event.CircuitEvent); ok {
|
|
ctx.Req.Equal("circuit", circuitEvent.Namespace)
|
|
ctx.Req.Equal("deleted", string(circuitEvent.EventType))
|
|
ctx.Req.Equal(clientIdentity.Id, circuitEvent.ClientId)
|
|
} else {
|
|
ctx.Req.Fail("unexpected event type: %v", reflect.TypeOf(evt))
|
|
}
|
|
}
|
|
}
|
|
|
|
func Test_ServiceBusEventLogger(t *testing.T) {
|
|
ctx := NewTestContext(t)
|
|
defer ctx.Teardown()
|
|
|
|
ctx.StartServer()
|
|
|
|
t.Run("servicebus event handler factory", func(t *testing.T) {
|
|
factory := events.ServiceBusEventLoggerFactory{}
|
|
|
|
// Test valid topic configuration
|
|
topicConfig := map[interface{}]interface{}{
|
|
"connectionString": "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=test-key",
|
|
"topic": "test-topic",
|
|
"format": "json",
|
|
"bufferSize": 100,
|
|
}
|
|
|
|
handler, err := factory.NewEventHandler(topicConfig)
|
|
ctx.Req.NoError(err)
|
|
ctx.Req.NotNil(handler)
|
|
|
|
// Test valid queue configuration
|
|
queueConfig := map[interface{}]interface{}{
|
|
"connectionString": "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=test-key",
|
|
"queue": "test-queue",
|
|
"format": "json",
|
|
}
|
|
|
|
handler, err = factory.NewEventHandler(queueConfig)
|
|
ctx.Req.NoError(err)
|
|
ctx.Req.NotNil(handler)
|
|
|
|
// Test missing connection string
|
|
invalidConfig := map[interface{}]interface{}{
|
|
"topic": "test-topic",
|
|
"format": "json",
|
|
}
|
|
|
|
_, err = factory.NewEventHandler(invalidConfig)
|
|
ctx.Req.Error(err)
|
|
ctx.Req.Contains(err.Error(), "unable to parse service bus config")
|
|
})
|
|
|
|
t.Run("servicebus configuration validation", func(t *testing.T) {
|
|
factory := events.ServiceBusEventLoggerFactory{}
|
|
|
|
// Test missing topic and queue
|
|
invalidConfig := map[interface{}]interface{}{
|
|
"connectionString": "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=test-key",
|
|
"format": "json",
|
|
}
|
|
|
|
_, err := factory.NewEventHandler(invalidConfig)
|
|
ctx.Req.Error(err)
|
|
ctx.Req.Contains(err.Error(), "unable to parse service bus config")
|
|
|
|
// Test invalid format
|
|
invalidFormatConfig := map[interface{}]interface{}{
|
|
"connectionString": "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=test-key",
|
|
"topic": "test-topic",
|
|
"format": "invalid",
|
|
}
|
|
|
|
_, err = factory.NewEventHandler(invalidFormatConfig)
|
|
ctx.Req.Error(err)
|
|
ctx.Req.Contains(err.Error(), "invalid 'format' for event log output file")
|
|
})
|
|
}
|