evan/conduit
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: evan:20c1388cf46e1f56f813ac419827fb30b81ec670
Branches Tags
evan:main
evan:0.0.1
..
compare: evan:de23b3e815a691f07c13ae28909c55ddd9f9d174
Branches Tags
evan:main
evan:0.0.1

No commits in common. "20c1388cf46e1f56f813ac419827fb30b81ec670" and "de23b3e815a691f07c13ae28909c55ddd9f9d174" have entirely different histories.
20c1388cf4 ... de23b3e815

5 changed files with 67 additions and 111 deletions
Show Stats Expand all files Collapse all files

2
client/client.go
View File

@ -41,5 +41,5 @@ func NewTunnel(cfg *config.ClientConfig) (*tunnel.Tunnel, error) {
return nil, fmt.Errorf("failed to connect: %v", err)
}
return tunnel.NewClientTunnel(cfg.TunnelName, cfg.TunnelTarget, serverURL, serverConn)
return tunnel.NewClientTunnel(cfg.TunnelName, cfg.TunnelTarget, serverConn)
}

51
pkg/maps/map.go
View File

@ -1,51 +0,0 @@
package maps
import (
"iter"
"sync"
)
type Map[K comparable, V any] struct {
items map[K]V
mu sync.RWMutex
}
func New[K comparable, V any]() *Map[K, V] {
return &Map[K, V]{items: make(map[K]V)}
}
func (m *Map[K, V]) Get(key K) (V, bool) {
m.mu.RLock()
defer m.mu.RUnlock()
v, ok := m.items[key]
return v, ok
}
func (m *Map[K, V]) Set(key K, value V) {
m.mu.Lock()
defer m.mu.Unlock()
m.items[key] = value
}
func (m *Map[K, V]) Delete(key K) {
m.mu.Lock()
defer m.mu.Unlock()
delete(m.items, key)
}
func (m *Map[K, V]) HasKey(key K) bool {
m.mu.RLock()
defer m.mu.RUnlock()
_, ok := m.items[key]
return ok
}
func (m *Map[K, V]) Entries() iter.Seq2[K, V] {
return func(yield func(K, V) bool) {
for k, v := range m.items {
if !yield(k, v) {
return
}
}
}
}

48
server/server.go
View File

@ -11,12 +11,12 @@ import (
"net/http"
"net/url"
"strings"
"sync"
"time"
"github.com/gorilla/websocket"
log "github.com/sirupsen/logrus"
"reichard.io/conduit/config"
"reichard.io/conduit/pkg/maps"
"reichard.io/conduit/tunnel"
)
@ -33,9 +33,10 @@ type TunnelInfo struct {
type Server struct {
host string
cfg *config.ServerConfig
mu sync.RWMutex
upgrader websocket.Upgrader
tunnels *maps.Map[string, *tunnel.Tunnel]
tunnels map[string]*tunnel.Tunnel
}
func NewServer(cfg *config.ServerConfig) (*Server, error) {
@ -49,7 +50,7 @@ func NewServer(cfg *config.ServerConfig) (*Server, error) {
return &Server{
cfg: cfg,
host: serverURL.Host,
tunnels: maps.New[string, *tunnel.Tunnel](),
tunnels: make(map[string]*tunnel.Tunnel),
upgrader: websocket.Upgrader{
CheckOrigin: func(r *http.Request) bool {
return true
@ -83,12 +84,14 @@ func (s *Server) Start() error {
func (s *Server) getInfo(w http.ResponseWriter, _ *http.Request) {
// Get Tunnels
var allTunnels []TunnelInfo
for t, c := range s.tunnels.Entries() {
s.mu.RLock()
for t, c := range s.tunnels {
allTunnels = append(allTunnels, TunnelInfo{
Name: t,
Target: c.Source(),
})
}
s.mu.RUnlock()
// Create Response
d, err := json.MarshalIndent(InfoResponse{
@ -135,31 +138,26 @@ func (s *Server) handleRawConnection(conn net.Conn) {
}
// Extract Subdomain
tunnelName := strings.TrimSuffix(strings.Replace(r.Host, s.host, "", 1), ".")
if strings.Count(tunnelName, ".") != 0 {
subdomain := strings.TrimSuffix(strings.Replace(r.Host, s.host, "", 1), ".")
if strings.Count(subdomain, ".") != 0 {
w.WriteHeader(http.StatusBadRequest)
_, _ = fmt.Fprintf(w, "cannot tunnel nested subdomains: %s", r.Host)
return
}
// Get True Host
remoteHost := conn.RemoteAddr().String()
if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
remoteHost = xff
}
r.RemoteAddr = remoteHost
// Handle Control Endpoints
if tunnelName == "" {
if subdomain == "" {
s.handleAsHTTP(w, r)
return
}
// Handle Tunnels
conduitTunnel, exists := s.tunnels.Get(tunnelName)
s.mu.RLock()
conduitTunnel, exists := s.tunnels[subdomain]
s.mu.RUnlock()
if !exists {
w.WriteHeader(http.StatusNotFound)
_, _ = fmt.Fprintf(w, "unknown tunnel: %s", tunnelName)
_, _ = fmt.Fprintf(w, "unknown tunnel: %s", subdomain)
return
}
@ -172,8 +170,8 @@ func (s *Server) handleRawConnection(conn net.Conn) {
return
}
log.Infof("tunnel %q connection from %s", tunnelName, r.RemoteAddr)
_ = conduitTunnel.StartStream(streamID, r.RemoteAddr)
log.Infof("relaying %s to tunnel", subdomain)
_ = conduitTunnel.StartStream(streamID)
}
func (s *Server) handleAsHTTP(w http.ResponseWriter, r *http.Request) {
@ -206,7 +204,7 @@ func (s *Server) createTunnel(w http.ResponseWriter, r *http.Request) {
}
// Validate Unique
if _, exists := s.tunnels.Get(tunnelName); exists {
if _, exists := s.tunnels[tunnelName]; exists {
w.WriteHeader(http.StatusConflict)
_, _ = w.Write([]byte("Tunnel already registered"))
return
@ -221,14 +219,18 @@ func (s *Server) createTunnel(w http.ResponseWriter, r *http.Request) {
// Create Tunnel
conduitTunnel := tunnel.NewServerTunnel(tunnelName, wsConn)
s.tunnels.Set(tunnelName, conduitTunnel)
log.Infof("tunnel %q created from %s", tunnelName, r.RemoteAddr)
s.mu.Lock()
s.tunnels[tunnelName] = conduitTunnel
s.mu.Unlock()
log.Infof("tunnel established: %s", tunnelName)
// Start Tunnel - This is blocking
conduitTunnel.Start()
// Cleanup Tunnel
s.tunnels.Delete(tunnelName)
s.mu.Lock()
delete(s.tunnels, tunnelName)
s.mu.Unlock()
_ = wsConn.Close()
log.Infof("tunnel %q closed from %s", tunnelName, r.RemoteAddr)
log.Infof("tunnel closed: %s", tunnelName)
}

62
tunnel/tunnel.go
View File

@ -9,7 +9,6 @@ import (
"github.com/gorilla/websocket"
log "github.com/sirupsen/logrus"
"reichard.io/conduit/pkg/maps"
"reichard.io/conduit/types"
)
@ -18,33 +17,27 @@ type ConnBuilder func() (conn io.ReadWriteCloser, err error)
func NewServerTunnel(name string, wsConn *websocket.Conn) *Tunnel {
return &Tunnel{
name: name,
streams: maps.New[string, io.ReadWriteCloser](),
wsConn: wsConn,
streams: make(map[string]io.ReadWriteCloser),
}
}
func NewClientTunnel(name, target string, serverURL *url.URL, wsConn *websocket.Conn) (*Tunnel, error) {
// Get Target URL
func NewClientTunnel(name, target string, wsConn *websocket.Conn) (*Tunnel, error) {
targetURL, err := url.Parse(target)
if err != nil {
return nil, err
}
// Derive Conduit URL
conduitURL := *serverURL
conduitURL.Host = name + "." + conduitURL.Host
// Get Connection Builder
var connBuilder ConnBuilder
switch targetURL.Scheme {
case "http", "https":
log.Infof("creating HTTP tunnel: %s -> %s", conduitURL.String(), target)
log.Infof("creating HTTP tunnel: %s -> %s", name, target)
connBuilder, err = HTTPConnectionBuilder(targetURL)
if err != nil {
return nil, err
}
default:
log.Infof("creating TCP tunnel: %s -> %s", conduitURL.String(), target)
log.Infof("creating TCP tunnel: %s -> %s", name, target)
connBuilder = func() (conn io.ReadWriteCloser, err error) {
return net.Dial("tcp", target)
}
@ -53,7 +46,7 @@ func NewClientTunnel(name, target string, serverURL *url.URL, wsConn *websocket.
return &Tunnel{
name: name,
wsConn: wsConn,
streams: maps.New[string, io.ReadWriteCloser](),
streams: make(map[string]io.ReadWriteCloser),
connBuilder: connBuilder,
}, nil
}
@ -61,10 +54,10 @@ func NewClientTunnel(name, target string, serverURL *url.URL, wsConn *websocket.
type Tunnel struct {
name string
wsConn *websocket.Conn
streams *maps.Map[string, io.ReadWriteCloser]
streams map[string]io.ReadWriteCloser
connBuilder ConnBuilder
mu sync.Mutex
wsMu, streamsMu sync.Mutex
}
func (t *Tunnel) Start() {
@ -102,7 +95,7 @@ func (t *Tunnel) initStreamConnection(streamID string) error {
return nil
}
if _, found := t.streams.Get(streamID); found {
if _, found := t.getStream(streamID); found {
return nil
}
@ -115,21 +108,24 @@ func (t *Tunnel) initStreamConnection(streamID string) error {
return err
}
go t.StartStream(streamID, "")
go t.StartStream(streamID)
return nil
}
func (t *Tunnel) AddStream(streamID string, conn io.ReadWriteCloser) error {
if t.streams.HasKey(streamID) {
t.streamsMu.Lock()
defer t.streamsMu.Unlock()
if _, found := t.streams[streamID]; found {
return fmt.Errorf("stream %s already exists", streamID)
}
t.streams.Set(streamID, conn)
t.streams[streamID] = conn
return nil
}
func (t *Tunnel) StartStream(streamID string, sourceAddr string) error {
func (t *Tunnel) StartStream(streamID string) error {
// Get Stream
conn, found := t.streams.Get(streamID)
conn, found := t.getStream(streamID)
if !found {
return fmt.Errorf("stream %s does not exist", streamID)
}
@ -139,7 +135,6 @@ func (t *Tunnel) StartStream(streamID string, sourceAddr string) error {
_ = t.sendWS(&types.Message{
Type: types.MessageTypeClose,
StreamID: streamID,
SourceAddr: sourceAddr,
})
t.CloseStream(streamID)
@ -155,9 +150,8 @@ func (t *Tunnel) StartStream(streamID string, sourceAddr string) error {
if err := t.sendWS(&types.Message{
Type: types.MessageTypeData,
StreamID: streamID,
Data: buffer[:n],
SourceAddr: sourceAddr,
StreamID: streamID,
}); err != nil {
return err
}
@ -166,7 +160,7 @@ func (t *Tunnel) StartStream(streamID string, sourceAddr string) error {
func (t *Tunnel) WriteStream(streamID string, data []byte) error {
// Get Stream
conn, found := t.streams.Get(streamID)
conn, found := t.getStream(streamID)
if !found {
return fmt.Errorf("stream %s does not exist", streamID)
}
@ -176,8 +170,10 @@ func (t *Tunnel) WriteStream(streamID string, data []byte) error {
}
func (t *Tunnel) CloseStream(streamID string) error {
if conn, ok := t.streams.Get(streamID); ok {
t.streams.Delete(streamID)
t.streamsMu.Lock()
defer t.streamsMu.Unlock()
if conn, ok := t.streams[streamID]; ok {
delete(t.streams, streamID)
return conn.Close()
}
return nil
@ -188,7 +184,17 @@ func (t *Tunnel) Source() string {
}
func (t *Tunnel) sendWS(msg *types.Message) error {
t.mu.Lock()
defer t.mu.Unlock()
t.wsMu.Lock()
defer t.wsMu.Unlock()
return t.wsConn.WriteJSON(msg)
}
func (t *Tunnel) getStream(streamID string) (io.ReadWriteCloser, bool) {
t.streamsMu.Lock()
defer t.streamsMu.Unlock()
if conn, ok := t.streams[streamID]; ok {
return conn, true
}
return nil, false
}

1
types/message.go
View File

@ -10,6 +10,5 @@ const (
type Message struct {
Type MessageType `json:"type"`
StreamID string `json:"stream_id"`
SourceAddr string `json:"source_addr"`
Data []byte `json:"data,omitempty"`
}