Merge pull request #368 from urnetwork/proxy-api

Proxy api

Author: xcolwell

connect/resident_proxy.go

@@ -124,7 +124,7 @@ func NewResidentProxyDevice(
 		deviceLocal.SetConnectLocation(initialDeviceState.Location)
 	}
 
-	tnet, err := proxy.CreateNetTUN(
+	tnet, err := proxy.CreateNetTun(
 		cancelCtx,
 		settings.Mtu,
 	)

http.go

@@ -3,6 +3,7 @@ package server
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
 	"encoding/json"
 	// "errors"
 	"fmt"
@@ -390,6 +391,7 @@ func HttpListenAndServeWithReusePort(ctx context.Context, addr string, handler h
 	if err != nil {
 		return err
 	}
+	defer listener.Close()
 
 	server := &http.Server{
 		Addr:         addr,
@@ -401,3 +403,27 @@ func HttpListenAndServeWithReusePort(ctx context.Context, addr string, handler h
 
 	return server.Serve(listener)
 }
+
+func HttpListenAndServeTlsWithReusePort(ctx context.Context, addr string, handler http.Handler, reusePort bool, httpServerOptions HttpServerOptions, tlsConfig *tls.Config) error {
+	listenConfig := net.ListenConfig{}
+	if reusePort {
+		listenConfig.Control = SoReusePort
+	}
+
+	listener, err := listenConfig.Listen(ctx, "tcp", addr)
+	if err != nil {
+		return err
+	}
+	defer listener.Close()
+
+	server := &http.Server{
+		Addr:         addr,
+		Handler:      handler,
+		TLSConfig:    tlsConfig,
+		ReadTimeout:  httpServerOptions.ReadTimeout,
+		WriteTimeout: httpServerOptions.WriteTimeout,
+		IdleTimeout:  httpServerOptions.IdleTimeout,
+	}
+
+	return server.ServeTLS(listener, "", "")
+}

model/network_client_model.go

@@ -126,12 +126,14 @@ type ProxyConfigResult struct {
 	SocksProxyUrl    string    `json:"socks_proxy_url"`
 	HttpProxyUrl     string    `json:"http_proxy_url"`
 	HttpsProxyUrl    string    `json:"https_proxy_url"`
+	ApiBaseUrl       string    `json:"api_base_url"`
 	AuthToken        string    `json:"auth_token"`
 	InstanceId       server.Id `json:"instance_id"`
 	ProxyHost        string    `json:"proxy_host"`
 	HttpProxyPort    int       `json:"http_proxy_port"`
 	HttpsProxyPort   int       `json:"https_proxy_port"`
 	SocksProxyPort   int       `json:"socks_proxy_port"`
+	ApiPort          int       `json:"api_port"`
 }
 
 type ProxyAuthResult struct {
@@ -286,6 +288,7 @@ func AuthNetworkClient(
 				socksProxyPort := 8080
 				httpProxyPort := 8081
 				httpsProxyPort := 8082
+				apiPort := 8083
 
 				proxyHost := fmt.Sprintf("%s.%s", "cosmic", server.RequireDomain())
 
@@ -314,16 +317,24 @@ func AuthNetworkClient(
 					)
 				}
 
+				apiBaseUrl := fmt.Sprintf(
+					"https://api.%s:%d",
+					proxyHost,
+					apiPort,
+				)
+
 				authClientResult.ProxyConfigResult = &ProxyConfigResult{
 					SocksProxyUrl:  socksProxyUrl,
 					HttpProxyUrl:   httpProxyUrl,
 					HttpsProxyUrl:  httpsProxyUrl,
+					ApiBaseUrl:     apiBaseUrl,
 					AuthToken:      strings.ToLower(signedProxyId),
 					InstanceId:     proxyDeviceConfig.InstanceId,
 					ProxyHost:      proxyHost,
 					SocksProxyPort: socksProxyPort,
 					HttpProxyPort:  httpProxyPort,
 					HttpsProxyPort: httpsProxyPort,
+					ApiPort:        apiPort,
 				}
 			} else {
 				authClientResult.Error = &AuthNetworkClientError{

proxy/main.go

@@ -2,13 +2,16 @@ package main
 
 import (
 	"context"
-	// "crypto/tls"
+	"crypto/tls"
 	"encoding/base64"
+	"encoding/json"
 	"fmt"
+	"io"
 	"net"
 	"net/http"
 	"net/netip"
 	"os"
+	"strconv"
 	"strings"
 	"syscall"
 	"time"
@@ -21,6 +24,7 @@ import (
 	"github.com/urnetwork/proxy"
 	"github.com/urnetwork/server"
 	"github.com/urnetwork/server/model"
+	"github.com/urnetwork/server/router"
 )
 
 // FIXME this is meant to be deployed with no lb and no containers
@@ -35,9 +39,11 @@ import (
 // -ldflags "-X main.Version=$WARP_VERSION-$WARP_VERSION_CODE"
 var Version string
 
+// FIXME warp port blocks
 const ListenSocksPort = 8080
 const ListenHttpPort = 8081
 const ListenHttpsPort = 8082
+const ListenApiPort = 8083
 
 func DefaultProxySettings() *ProxySettings {
 	return &ProxySettings{
@@ -89,7 +95,15 @@ func main() {
 		panic(err)
 	}
 
-	glog.Infof("Listen socks5 (:%d), http (:%d), https (:%d)", ListenSocksPort, ListenHttpPort, ListenHttpsPort)
+	glog.Infof("Listen api (:%d), socks5 (:%d), http (:%d), https (:%d)", ListenApiPort, ListenSocksPort, ListenHttpPort, ListenHttpsPort)
+
+	newApiServer(
+		ctx,
+		cancel,
+		proxyDeviceManager,
+		transportTls,
+		settings,
+	)
 
 	newSocks5Server(
 		ctx,
@@ -242,31 +256,8 @@ func (self *httpServer) run() {
 			}
 		}
 
-		headerAuth := r.Header.Get("Proxy-Authorization")
-
-		bearerPrefix := "bearer "
-		basicPrefix := "basic "
-
-		if len(bearerPrefix) < len(headerAuth) && strings.ToLower(headerAuth[:len(bearerPrefix)]) == bearerPrefix {
-			signedProxyId := headerAuth[len(bearerPrefix):]
-			proxyId, err := model.ParseSignedProxyId(signedProxyId)
-			if err == nil {
-				return proxyId, nil
-			}
-		} else if len(basicPrefix) < len(headerAuth) && strings.ToLower(headerAuth[:len(basicPrefix)]) == basicPrefix {
-			// user:pass
-			combinedSignedProxyId, err := base64.StdEncoding.DecodeString(headerAuth[len(basicPrefix):])
-			if err != nil {
-				return server.Id{}, err
-			}
-			signedProxyId := strings.SplitN(string(combinedSignedProxyId), ":", 2)[0]
-			proxyId, err := model.ParseSignedProxyId(signedProxyId)
-			if err == nil {
-				return proxyId, nil
-			}
-		}
-
-		return server.Id{}, fmt.Errorf("Not authorized")
+		authHeader := r.Header.Get("Proxy-Authorization")
+		return authHeaderProxyId(authHeader)
 	}
 
 	connectDial := func(r *http.Request, network string, addr string) (net.Conn, error) {
@@ -333,3 +324,140 @@ func (self *httpServer) run() {
 	case <-self.ctx.Done():
 	}
 }
+
+type apiServer struct {
+	ctx                context.Context
+	cancel             context.CancelFunc
+	proxyDeviceManager *ProxyDeviceManager
+	transportTls       *server.TransportTls
+	settings           *ProxySettings
+}
+
+func newApiServer(
+	ctx context.Context,
+	cancel context.CancelFunc,
+	proxyDeviceManager *ProxyDeviceManager,
+	transportTls *server.TransportTls,
+	settings *ProxySettings,
+) *apiServer {
+	s := &apiServer{
+		ctx:                ctx,
+		cancel:             cancel,
+		proxyDeviceManager: proxyDeviceManager,
+		transportTls:       transportTls,
+		settings:           settings,
+	}
+
+	go server.HandleError(s.run, cancel)
+
+	return s
+}
+
+func (self *apiServer) run() {
+	defer self.cancel()
+
+	routes := []*router.Route{
+		router.NewRoute("POST", "/warmup", self.HandleWarmup),
+	}
+
+	reusePort := false
+
+	httpServerOptions := server.HttpServerOptions{
+		ReadTimeout:  15 * time.Second,
+		WriteTimeout: 30 * time.Second,
+		IdleTimeout:  5 * time.Minute,
+	}
+
+	tlsConfig := &tls.Config{
+		GetConfigForClient: self.transportTls.GetTlsConfigForClient,
+	}
+
+	err := server.HttpListenAndServeTlsWithReusePort(
+		self.ctx,
+		net.JoinHostPort("", strconv.Itoa(ListenApiPort)),
+		router.NewRouter(self.ctx, routes),
+		reusePort,
+		httpServerOptions,
+		tlsConfig,
+	)
+	if err != nil {
+		panic(err)
+	}
+}
+
+type WarmupRequest struct {
+	TimeoutSeconds int `json:"timeout_seconds,omitempty"`
+}
+
+type WarmupResponse struct {
+	Ready bool `json:"ready"`
+}
+
+func (self *apiServer) HandleWarmup(w http.ResponseWriter, r *http.Request) {
+	authHeader := r.Header.Get("Authorization")
+	proxyId, err := authHeaderProxyId(authHeader)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusUnauthorized)
+		return
+	}
+
+	var warmupRequest WarmupRequest
+
+	defer r.Body.Close()
+	bodyBytes, err := io.ReadAll(r.Body)
+
+	if 0 < len(bodyBytes) {
+		err = json.Unmarshal(bodyBytes, &warmupRequest)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusInternalServerError)
+			return
+		}
+	}
+	// else use the default object
+
+	proxyDevice, err := self.proxyDeviceManager.OpenProxyDevice(proxyId)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	timeout := time.Duration(warmupRequest.TimeoutSeconds) * time.Second
+	ready := proxyDevice.WaitForReady(r.Context(), timeout)
+
+	warmupResponse := &WarmupResponse{
+		Ready: ready,
+	}
+
+	out, err := json.Marshal(warmupResponse)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	w.Write(out)
+}
+
+func authHeaderProxyId(authHeader string) (server.Id, error) {
+	bearerPrefix := "bearer "
+	basicPrefix := "basic "
+
+	if len(bearerPrefix) < len(authHeader) && strings.ToLower(authHeader[:len(bearerPrefix)]) == bearerPrefix {
+		signedProxyId := authHeader[len(bearerPrefix):]
+		proxyId, err := model.ParseSignedProxyId(signedProxyId)
+		if err == nil {
+			return proxyId, nil
+		}
+	} else if len(basicPrefix) < len(authHeader) && strings.ToLower(authHeader[:len(basicPrefix)]) == basicPrefix {
+		// user:pass
+		combinedSignedProxyId, err := base64.StdEncoding.DecodeString(authHeader[len(basicPrefix):])
+		if err != nil {
+			return server.Id{}, err
+		}
+		signedProxyId := strings.SplitN(string(combinedSignedProxyId), ":", 2)[0]
+		proxyId, err := model.ParseSignedProxyId(signedProxyId)
+		if err == nil {
+			return proxyId, nil
+		}
+	}
+
+	return server.Id{}, fmt.Errorf("Not authorized")
+}

proxy/proxy_device.go

@@ -276,6 +276,59 @@ func (self *ProxyDevice) Tun() *proxy.Net {
 	return self.tnet
 }
 
+func (self *ProxyDevice) WaitForReady(ctx context.Context, timeout time.Duration) bool {
+	readyCtx, readyCancel := context.WithCancel(self.ctx)
+	defer readyCancel()
+	go server.HandleError(func() {
+		defer readyCancel()
+		select {
+		case <-self.ctx.Done():
+		case <-ctx.Done():
+		}
+	})
+
+	windowStatus := self.deviceLocal.GetWindowStatus()
+	if windowStatus.MinSatisfied {
+		return true
+	}
+
+	if timeout == 0 {
+		return false
+	}
+
+	sub := self.deviceLocal.AddWindowStatusChangeListener(&windowStatusChangeListener{
+		callback: func(windowStatus *sdk.WindowStatus) {
+			if windowStatus.MinSatisfied {
+				readyCancel()
+			}
+		},
+	})
+	defer sub.Close()
+
+	if 0 < timeout {
+		select {
+		case <-readyCtx.Done():
+			return true
+		case <-time.After(timeout):
+			return false
+		}
+	} else {
+		select {
+		case <-readyCtx.Done():
+			return true
+		}
+	}
+}
+
+// conforms to `sdk.WindowStatusChangeListener`
+type windowStatusChangeListener struct {
+	callback func(*sdk.WindowStatus)
+}
+
+func (self *windowStatusChangeListener) WindowStatusChanged(windowStatus *sdk.WindowStatus) {
+	self.callback(windowStatus)
+}
+
 func (self *ProxyDevice) UpdateActivity() bool {
 	self.stateLock.Lock()
 	defer self.stateLock.Unlock()