CyberCoveLLC/threshold
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

repo: Return connection informat in Session response object (#1690)

Browse Source 
Add connections to the Session object

* db: creat new session_list view

* repo: session use new session_list view
This commit is contained in:
Thor authored and GitHub committed 2021-11-15 15:26:08 -06:00
1 parent 8cd8453113
commit ad31b6d6c3
6 files changed
+164 -23
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/db/schema/migrations/oss/postgres/20/01_session_list_view.up.sql
+50
View File
@@ -0,0 +1,50 @@
begin;
-- Replaces the view created in 1/01 to include connections
drop view session_with_state;
create view session_list as
select
s.public_id,
s.user_id,
s.host_id,
s.server_id,
s.server_type,
s.target_id,
s.host_set_id,
s.auth_token_id,
s.scope_id,
s.certificate,
s.expiration_time,
s.connection_limit,
s.tofu_token,
s.key_id,
s.termination_reason,
s.version,
s.create_time,
s.update_time,
s.endpoint,
s.worker_filter,
ss.state,
ss.previous_end_time,
ss.start_time,
ss.end_time,
sc.public_id as connection_id,
sc.client_tcp_address,
sc.client_tcp_port,
sc.endpoint_tcp_address,
sc.endpoint_tcp_port,
sc.bytes_up,
sc.bytes_down,
sc.closed_reason
from
session s
join
session_state ss
on
s.public_id = ss.session_id
left join
session_connection sc
on
s.public_id = sc.session_id;
commit;
internal/session/query.go
+1 -1
View File
@@ -155,7 +155,7 @@ from
select *
from
(select public_id from session %s) s,
session_with_state ss
session_list ss
where
s.public_id = ss.public_id
%s
internal/session/repository.go
+47 -10
View File
@@ -5,6 +5,7 @@ import (
"sort"
"github.com/hashicorp/boundary/internal/db"
"github.com/hashicorp/boundary/internal/db/timestamp"
"github.com/hashicorp/boundary/internal/errors"
"github.com/hashicorp/boundary/internal/kms"
)
@@ -74,19 +75,31 @@ func (r *Repository) list(ctx context.Context, resources interface{}, where stri
return nil
}
func (r *Repository) convertToSessions(ctx context.Context, sessionsWithState []*sessionView, opt ...Option) ([]*Session, error) {
func (r *Repository) convertToSessions(ctx context.Context, sessionList []*sessionListView, opt ...Option) ([]*Session, error) {
const op = "session.(Repository).convertToSessions"
opts := getOpts(opt...)
if len(sessionsWithState) == 0 {
if len(sessionList) == 0 {
return nil, nil
}
sessions := []*Session{}
// deduplication map of connections by connection_id
connections := map[string]*Connection{}
// deduplication map of states by end_time
states := map[*timestamp.Timestamp]*State{}
var prevSessionId string
var workingSession *Session
for _, sv := range sessionsWithState {
for _, sv := range sessionList {
if sv.PublicId != prevSessionId {
if prevSessionId != "" {
for _, c := range connections {
workingSession.Connections = append(workingSession.Connections, c)
}
connections = map[string]*Connection{}
for _, s := range states {
workingSession.States = append(workingSession.States, s)
}
states = map[*timestamp.Timestamp]*State{}
sort.Slice(workingSession.States, func(i, j int) bool {
return workingSession.States[i].StartTime.GetTimestamp().AsTime().After(workingSession.States[j].StartTime.GetTimestamp().AsTime())
})
@@ -134,18 +147,42 @@ func (r *Repository) convertToSessions(ctx context.Context, sessionsWithState []
}
}
state := &State{
SessionId: sv.PublicId,
Status: Status(sv.Status),
PreviousEndTime: sv.PreviousEndTime,
StartTime: sv.StartTime,
EndTime: sv.EndTime,
if _, ok := states[sv.EndTime]; !ok {
states[sv.EndTime] = &State{
SessionId: sv.PublicId,
Status: Status(sv.Status),
PreviousEndTime: sv.PreviousEndTime,
StartTime: sv.StartTime,
EndTime: sv.EndTime,
}
}
workingSession.States = append(workingSession.States, state)
if sv.ConnectionId != "" {
if _, ok := connections[sv.ConnectionId]; !ok {
connections[sv.ConnectionId] = &Connection{
PublicId: sv.ConnectionId,
SessionId: sv.PublicId,
ClientTcpAddress: sv.ClientTcpAddress,
ClientTcpPort: sv.ClientTcpPort,
EndpointTcpAddress: sv.EndpointTcpAddress,
EndpointTcpPort: sv.EndpointTcpPort,
BytesUp: sv.BytesUp,
BytesDown: sv.BytesDown,
ClosedReason: sv.ClosedReason,
}
}
}
}
for _, s := range states {
workingSession.States = append(workingSession.States, s)
}
sort.Slice(workingSession.States, func(i, j int) bool {
return workingSession.States[i].StartTime.GetTimestamp().AsTime().After(workingSession.States[j].StartTime.GetTimestamp().AsTime())
})
for _, c := range connections {
workingSession.Connections = append(workingSession.Connections, c)
}
sessions = append(sessions, workingSession)
return sessions, nil
}
internal/session/repository_session.go
+24 -4
View File
@@ -166,6 +166,12 @@ func (r *Repository) LookupSession(ctx context.Context, sessionId string, _ ...O
if len(creds) > 0 {
session.DynamicCredentials = creds
}
connections, err := fetchConnections(ctx, read, sessionId, db.WithOrder("create_time desc"))
if err != nil {
return errors.Wrap(ctx, err, op)
}
session.Connections = connections
return nil
},
)
@@ -249,6 +255,8 @@ func (r *Repository) ListSessions(ctx context.Context, opt ...Option) ([]*Sessio
case db.AscendingOrderBy:
withOrder = "order by create_time asc"
case db.DescendingOrderBy:
fallthrough
default:
withOrder = "order by create_time"
}
@@ -265,15 +273,15 @@ func (r *Repository) ListSessions(ctx context.Context, opt ...Option) ([]*Sessio
}
defer rows.Close()
var sessionsWithState []*sessionView
var sessionsList []*sessionListView
for rows.Next() {
var s sessionView
var s sessionListView
if err := r.reader.ScanRows(rows, &s); err != nil {
return nil, errors.Wrap(ctx, err, op, errors.WithMsg("scan row failed"))
}
sessionsWithState = append(sessionsWithState, &s)
sessionsList = append(sessionsList, &s)
}
sessions, err := r.convertToSessions(ctx, sessionsWithState, withListingConvert(true))
sessions, err := r.convertToSessions(ctx, sessionsList, withListingConvert(true))
if err != nil {
return nil, errors.Wrap(ctx, err, op)
}
@@ -790,3 +798,15 @@ func fetchStates(ctx context.Context, r db.Reader, sessionId string, opt ...db.O
}
return states, nil
}
func fetchConnections(ctx context.Context, r db.Reader, sessionId string, opt ...db.Option) ([]*Connection, error) {
const op = "session.fetchConnections"
var connections []*Connection
if err := r.SearchWhere(ctx, &connections, "session_id = ?", []interface{}{sessionId}, opt...); err != nil {
return nil, errors.Wrap(ctx, err, op)
}
if len(connections) == 0 {
return nil, nil
}
return connections, nil
}
internal/session/repository_session_test.go
+26 -5
View File
@@ -48,11 +48,12 @@ func TestRepository_ListSession(t *testing.T) {
opt []Option
}
tests := []struct {
name string
createCnt int
args args
wantCnt int
wantErr bool
name string
createCnt int
args args
wantCnt int
wantErr bool
withConnections int
}{
{
name: "no-limit",
@@ -97,6 +98,14 @@ func TestRepository_ListSession(t *testing.T) {
wantCnt: 0,
wantErr: false,
},
{
name: "multiple-connections",
createCnt: repo.defaultLimit + 1,
args: args{},
wantCnt: repo.defaultLimit,
wantErr: false,
withConnections: 3,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
@@ -107,6 +116,9 @@ func TestRepository_ListSession(t *testing.T) {
s := TestSession(t, conn, wrapper, composedOf)
_ = TestState(t, conn, s.PublicId, StatusActive)
testSessions = append(testSessions, s)
for i := 0; i < tt.withConnections; i++ {
_ = TestConnection(t, conn, s.PublicId, "127.0.0.1", 22, "127.0.0.2", 23)
}
}
assert.Equal(tt.createCnt, len(testSessions))
got, err := repo.ListSessions(context.Background(), tt.args.opt...)
@@ -116,6 +128,15 @@ func TestRepository_ListSession(t *testing.T) {
}
require.NoError(err)
assert.Equal(tt.wantCnt, len(got))
for i := 0; i < len(got); i++ {
assert.Equal(tt.withConnections, len(got[i].Connections))
for _, c := range got[i].Connections {
assert.Equal("127.0.0.1", c.ClientTcpAddress)
assert.Equal(uint32(22), c.ClientTcpPort)
assert.Equal("127.0.0.2", c.EndpointTcpAddress)
assert.Equal(uint32(23), c.EndpointTcpPort)
}
}
if tt.wantCnt > 0 {
assert.Equal(StatusActive, got[0].States[0].Status)
assert.Equal(StatusPending, got[0].States[1].Status)
internal/session/session.go
+16 -3
View File
@@ -112,6 +112,9 @@ type Session struct {
// DynamicCredentials for the session.
DynamicCredentials []*DynamicCredential `gorm:"-"`
// Connections for the session are for read only and are ignored during write operations
Connections []*Connection `gorm:"-"`
tableName string `gorm:"-"`
}
@@ -401,7 +404,7 @@ func (s *Session) decrypt(ctx context.Context, cipher wrapping.Wrapper) error {
return nil
}
type sessionView struct {
type sessionListView struct {
// Session fields
PublicId string `json:"public_id,omitempty" gorm:"primary_key"`
UserId string `json:"user_id,omitempty" gorm:"default:null"`
@@ -429,9 +432,19 @@ type sessionView struct {
PreviousEndTime *timestamp.Timestamp `json:"previous_end_time,omitempty" gorm:"default:current_timestamp"`
StartTime *timestamp.Timestamp `json:"start_time,omitempty" gorm:"default:current_timestamp;primary_key"`
EndTime *timestamp.Timestamp `json:"end_time,omitempty" gorm:"default:current_timestamp"`
// Connection fields
ConnectionId string `json:"connection_id,omitempty" gorm:"default:null"`
ClientTcpAddress string `json:"client_tcp_address,omitempty" gorm:"default:null"`
ClientTcpPort uint32 `json:"client_tcp_port,omitempty" gorm:"default:null"`
EndpointTcpAddress string `json:"endpoint_tcp_address,omitempty" gorm:"default:null"`
EndpointTcpPort uint32 `json:"endpoint_tcp_port,omitempty" gorm:"default:null"`
BytesUp uint64 `json:"bytes_up,omitempty" gorm:"default:null"`
BytesDown uint64 `json:"bytes_down,omitempty" gorm:"default:null"`
ClosedReason string `json:"closed_reason,omitempty" gorm:"default:null"`
}
// TableName returns the tablename to override the default gorm table name
func (s *sessionView) TableName() string {
return "session_with_state"
func (s *sessionListView) TableName() string {
return "session_list"
}