[feat](protocol, client, server): replace DTLS with gRPC for tunnel implementation

- Introduced gRPC-based tunnel design for bi-directional communication, replacing legacy DTLS transport.
- Added `HopGateTunnel` gRPC service with client and server logic for `OpenTunnel` stream handling.
- Updated client to use gRPC tunnel exclusively, including experimental entry point for stream-based HTTP proxying.
- Removed DTLS-specific client, server, and related dependencies (`pion/dtls`).
- Adjusted `cmd/server` to route gRPC and HTTP/HTTPS traffic dynamically on shared ports.
This commit is contained in:
dalbodeule
2025-12-11 16:48:17 +09:00
parent 17839def69
commit 64f730d2df
8 changed files with 463 additions and 392 deletions

View File

@@ -5,14 +5,17 @@ import (
"crypto/tls"
"crypto/x509"
"flag"
"io"
"net"
"os"
"strings"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
"github.com/dalbodeule/hop-gate/internal/config"
"github.com/dalbodeule/hop-gate/internal/dtls"
"github.com/dalbodeule/hop-gate/internal/logging"
"github.com/dalbodeule/hop-gate/internal/proxy"
protocolpb "github.com/dalbodeule/hop-gate/internal/protocol/pb"
)
// version 은 빌드 시 -ldflags "-X main.version=xxxxxxx" 로 덮어쓰이는 필드입니다.
@@ -48,6 +51,148 @@ func firstNonEmpty(values ...string) string {
return ""
}
// runGRPCTunnelClient 는 gRPC 기반 터널을 사용하는 실험적 클라이언트 진입점입니다. (ko)
// runGRPCTunnelClient is an experimental entrypoint for a gRPC-based tunnel client. (en)
func runGRPCTunnelClient(ctx context.Context, logger logging.Logger, finalCfg *config.ClientConfig) error {
// TLS 설정은 기존 DTLS 클라이언트와 동일한 정책을 사용합니다. (ko)
// TLS configuration mirrors the existing DTLS client policy. (en)
var tlsCfg *tls.Config
if finalCfg.Debug {
tlsCfg = &tls.Config{
InsecureSkipVerify: true,
MinVersion: tls.VersionTLS12,
}
} else {
rootCAs, err := x509.SystemCertPool()
if err != nil || rootCAs == nil {
rootCAs = x509.NewCertPool()
}
tlsCfg = &tls.Config{
RootCAs: rootCAs,
MinVersion: tls.VersionTLS12,
}
}
// finalCfg.ServerAddr 가 "host:port" 형태이므로, SNI 에는 DNS(host) 부분만 넣어야 한다.
host := finalCfg.ServerAddr
if h, _, err := net.SplitHostPort(finalCfg.ServerAddr); err == nil && strings.TrimSpace(h) != "" {
host = h
}
tlsCfg.ServerName = host
creds := credentials.NewTLS(tlsCfg)
log := logger.With(logging.Fields{
"component": "grpc_tunnel_client",
"server_addr": finalCfg.ServerAddr,
"domain": finalCfg.Domain,
"local_target": finalCfg.LocalTarget,
})
log.Info("dialing grpc tunnel", nil)
conn, err := grpc.DialContext(ctx, finalCfg.ServerAddr, grpc.WithTransportCredentials(creds), grpc.WithBlock())
if err != nil {
log.Error("failed to dial grpc tunnel server", logging.Fields{
"error": err.Error(),
})
return err
}
defer conn.Close()
client := protocolpb.NewHopGateTunnelClient(conn)
stream, err := client.OpenTunnel(ctx)
if err != nil {
log.Error("failed to open grpc tunnel stream", logging.Fields{
"error": err.Error(),
})
return err
}
log.Info("grpc tunnel stream opened", nil)
// 초기 핸드셰이크: 도메인, API 키, 로컬 타깃 정보를 StreamOpen 헤더로 전송합니다. (ko)
// Initial handshake: send domain, API key, and local target via StreamOpen headers. (en)
headers := map[string]*protocolpb.HeaderValues{
"X-HopGate-Domain": {Values: []string{finalCfg.Domain}},
"X-HopGate-API-Key": {Values: []string{finalCfg.ClientAPIKey}},
"X-HopGate-Local-Target": {Values: []string{finalCfg.LocalTarget}},
}
open := &protocolpb.StreamOpen{
Id: "control-0",
ServiceName: "control",
TargetAddr: "",
Header: headers,
}
env := &protocolpb.Envelope{
Payload: &protocolpb.Envelope_StreamOpen{
StreamOpen: open,
},
}
if err := stream.Send(env); err != nil {
log.Error("failed to send initial stream_open handshake", logging.Fields{
"error": err.Error(),
})
return err
}
log.Info("sent initial stream_open handshake on grpc tunnel", logging.Fields{
"domain": finalCfg.Domain,
"local_target": finalCfg.LocalTarget,
"api_key_mask": maskAPIKey(finalCfg.ClientAPIKey),
})
// 수신 루프: 현재는 수신된 Envelope 의 타입만 로그에 남기고 종료하지 않습니다. (ko)
// Receive loop: currently only logs envelope payload types and keeps the tunnel open. (en)
for {
if ctx.Err() != nil {
log.Info("context cancelled, closing grpc tunnel client", logging.Fields{
"error": ctx.Err().Error(),
})
return ctx.Err()
}
in, err := stream.Recv()
if err != nil {
if err == io.EOF {
log.Info("grpc tunnel stream closed by server", nil)
return nil
}
log.Error("grpc tunnel receive error", logging.Fields{
"error": err.Error(),
})
return err
}
payloadType := "unknown"
switch in.Payload.(type) {
case *protocolpb.Envelope_HttpRequest:
payloadType = "http_request"
case *protocolpb.Envelope_HttpResponse:
payloadType = "http_response"
case *protocolpb.Envelope_StreamOpen:
payloadType = "stream_open"
case *protocolpb.Envelope_StreamData:
payloadType = "stream_data"
case *protocolpb.Envelope_StreamClose:
payloadType = "stream_close"
case *protocolpb.Envelope_StreamAck:
payloadType = "stream_ack"
}
log.Info("received envelope on grpc tunnel client", logging.Fields{
"payload_type": payloadType,
})
// 이후 단계에서 여기서 HTTP 프록시와의 연동(요청/응답 처리)을 구현할 예정입니다. (ko)
// Future 3.3 work will hook HTTP proxy logic here. (en)
}
}
func main() {
logger := logging.NewStdJSONLogger("client")
@@ -87,7 +232,7 @@ func main() {
})
// CLI 인자 정의 (env 보다 우선 적용됨)
serverAddrFlag := flag.String("server-addr", "", "DTLS server address (host:port)")
serverAddrFlag := flag.String("server-addr", "", "HopGate server address (host:port)")
domainFlag := flag.String("domain", "", "registered domain (e.g. api.example.com)")
apiKeyFlag := flag.String("api-key", "", "client API key for the domain (64 chars)")
localTargetFlag := flag.String("local-target", "", "local HTTP target (host:port), e.g. 127.0.0.1:8080")
@@ -136,78 +281,16 @@ func main() {
"debug": finalCfg.Debug,
})
// 4. DTLS 클라이언트 연결 및 핸드셰이크
ctx := context.Background()
// 디버그 모드에서는 서버 인증서 검증을 스킵(InsecureSkipVerify=true) 하여
// self-signed 테스트 인증서도 신뢰하도록 합니다.
// 운영 환경에서는 Debug=false 로 두고, 올바른 RootCAs / ServerName 을 갖는 tls.Config 를 사용해야 합니다.
var tlsCfg *tls.Config
if finalCfg.Debug {
tlsCfg = &tls.Config{
InsecureSkipVerify: true,
MinVersion: tls.VersionTLS12,
}
} else {
// 운영 모드: 시스템 루트 CA + SNI(ServerName)에 서버 도메인 설정
rootCAs, err := x509.SystemCertPool()
if err != nil || rootCAs == nil {
rootCAs = x509.NewCertPool()
}
tlsCfg = &tls.Config{
RootCAs: rootCAs,
MinVersion: tls.VersionTLS12,
}
}
// DTLS 서버 측은 SNI(ServerName)가 HOP_SERVER_DOMAIN(cfg.Domain)과 일치하는지 검사하므로,
// 클라이언트 TLS 설정에도 반드시 도메인을 설정해준다.
//
// finalCfg.ServerAddr 가 "host:port" 형태이므로, SNI 에는 DNS(host) 부분만 넣어야 한다.
host := finalCfg.ServerAddr
if h, _, err := net.SplitHostPort(finalCfg.ServerAddr); err == nil && strings.TrimSpace(h) != "" {
host = h
}
tlsCfg.ServerName = host
client := dtls.NewPionClient(dtls.PionClientConfig{
Addr: finalCfg.ServerAddr,
TLSConfig: tlsCfg,
})
sess, err := client.Connect()
if err != nil {
logger.Error("failed to establish dtls session", logging.Fields{
"error": err.Error(),
})
os.Exit(1)
}
defer sess.Close()
hsRes, err := dtls.PerformClientHandshake(ctx, sess, logger, finalCfg.Domain, finalCfg.ClientAPIKey, finalCfg.LocalTarget)
if err != nil {
logger.Error("dtls handshake failed", logging.Fields{
// 현재 클라이언트는 DTLS 레이어 없이 gRPC 터널만을 사용합니다. (ko)
// The client now uses only the gRPC tunnel, without any DTLS layer. (en)
if err := runGRPCTunnelClient(ctx, logger, finalCfg); err != nil {
logger.Error("grpc tunnel client exited with error", logging.Fields{
"error": err.Error(),
})
os.Exit(1)
}
logger.Info("dtls handshake completed", logging.Fields{
"domain": hsRes.Domain,
"local_target": finalCfg.LocalTarget,
})
// 5. DTLS 세션 위에서 서버 요청을 처리하는 클라이언트 프록시 루프 시작
clientProxy := proxy.NewClientProxy(logger, finalCfg.LocalTarget)
logger.Info("starting client proxy loop", logging.Fields{
"local_target": finalCfg.LocalTarget,
})
if err := clientProxy.StartLoop(ctx, sess); err != nil {
logger.Error("client proxy loop exited with error", logging.Fields{
"error": err.Error(),
})
os.Exit(1)
}
logger.Info("client proxy loop exited normally", nil)
logger.Info("grpc tunnel client exited normally", nil)
}

View File

@@ -19,6 +19,8 @@ import (
"time"
"github.com/prometheus/client_golang/prometheus/promhttp"
"google.golang.org/grpc"
"google.golang.org/grpc/peer"
"github.com/dalbodeule/hop-gate/internal/acme"
"github.com/dalbodeule/hop-gate/internal/admin"
@@ -28,6 +30,7 @@ import (
"github.com/dalbodeule/hop-gate/internal/logging"
"github.com/dalbodeule/hop-gate/internal/observability"
"github.com/dalbodeule/hop-gate/internal/protocol"
protocolpb "github.com/dalbodeule/hop-gate/internal/protocol/pb"
"github.com/dalbodeule/hop-gate/internal/store"
)
@@ -795,6 +798,8 @@ var (
sessionsByDomain = make(map[string]*dtlsSessionWrapper)
)
// statusRecorder 는 HTTP 응답 상태 코드를 캡처하기 위한 래퍼입니다.
// Prometheus 메트릭에서 status 라벨을 기록하는 데 사용합니다.
// statusRecorder 는 HTTP 응답 상태 코드를 캡처하기 위한 래퍼입니다.
// Prometheus 메트릭에서 status 라벨을 기록하는 데 사용합니다.
type statusRecorder struct {
@@ -807,6 +812,82 @@ func (w *statusRecorder) WriteHeader(code int) {
w.ResponseWriter.WriteHeader(code)
}
// grpcTunnelServer 는 HopGate gRPC 터널 서비스(HopGateTunnel)의 서버 구현체입니다. (ko)
// grpcTunnelServer implements the HopGateTunnel gRPC service on the server side. (en)
type grpcTunnelServer struct {
protocolpb.UnimplementedHopGateTunnelServer
logger logging.Logger
}
// newGRPCTunnelServer 는 gRPC 터널 서버 구현체를 생성합니다. (ko)
// newGRPCTunnelServer constructs a new gRPC tunnel server implementation. (en)
func newGRPCTunnelServer(logger logging.Logger) *grpcTunnelServer {
return &grpcTunnelServer{
logger: logger.With(logging.Fields{
"component": "grpc_tunnel",
}),
}
}
// OpenTunnel 은 클라이언트와 서버 간 장기 유지 bi-directional gRPC 스트림을 처리합니다. (ko)
// OpenTunnel handles the long-lived bi-directional gRPC stream between the
// server and a HopGate client. At this stage, it only logs incoming envelopes
// and does not yet integrate with the HTTP proxy layer. (en)
func (s *grpcTunnelServer) OpenTunnel(stream protocolpb.HopGateTunnel_OpenTunnelServer) error {
ctx := stream.Context()
// 원격 주소가 있으면 로그 필드에 추가합니다. (ko)
// Attach remote address from the peer info to log fields when available. (en)
fields := logging.Fields{}
if p, ok := peer.FromContext(ctx); ok && p.Addr != nil {
fields["remote_addr"] = p.Addr.String()
}
log := s.logger.With(fields)
log.Info("grpc tunnel opened", nil)
defer log.Info("grpc tunnel closed", nil)
for {
env, err := stream.Recv()
if err != nil {
if err == io.EOF {
// 클라이언트가 정상적으로 스트림을 종료한 경우. (ko)
// Client closed the stream normally. (en)
return nil
}
log.Error("grpc tunnel receive error", logging.Fields{
"error": err.Error(),
})
return err
}
// 현재 단계에서는 수신된 Envelope 의 payload 타입만 로그에 남기고,
// 실제 HTTP 프록시 연동은 후속 3.3 작업에서 구현합니다. (ko)
// At this stage we only log the envelope payload type; HTTP proxy
// integration will be implemented in later 3.3 steps. (en)
payloadType := "unknown"
switch env.Payload.(type) {
case *protocolpb.Envelope_HttpRequest:
payloadType = "http_request"
case *protocolpb.Envelope_HttpResponse:
payloadType = "http_response"
case *protocolpb.Envelope_StreamOpen:
payloadType = "stream_open"
case *protocolpb.Envelope_StreamData:
payloadType = "stream_data"
case *protocolpb.Envelope_StreamClose:
payloadType = "stream_close"
case *protocolpb.Envelope_StreamAck:
payloadType = "stream_ack"
}
log.Info("received envelope on grpc tunnel", logging.Fields{
"payload_type": payloadType,
})
}
}
// hopGateOwnedHeaders 는 HopGate 서버가 스스로 관리하는 응답 헤더 목록입니다. (ko)
// hopGateOwnedHeaders lists response headers that are owned by the HopGate server. (en)
var hopGateOwnedHeaders = map[string]struct{}{
@@ -887,6 +968,22 @@ func hostDomainHandler(allowedDomain string, logger logging.Logger, next http.Ha
})
}
// grpcOrHTTPHandler 는 단일 HTTPS 포트에서 gRPC(OpenTunnel)와 일반 HTTP 요청을
// Content-Type 및 프로토콜(HTTP/2) 기준으로 라우팅하는 헬퍼입니다. (ko)
// grpcOrHTTPHandler routes between gRPC (OpenTunnel) and regular HTTP handlers
// on a single HTTPS port, based on Content-Type and protocol (HTTP/2). (en)
func grpcOrHTTPHandler(grpcServer *grpc.Server, httpHandler http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// gRPC 요청은 HTTP/2 + Content-Type: application/grpc 조합으로 들어옵니다. (ko)
// gRPC requests arrive as HTTP/2 with Content-Type: application/grpc. (en)
if r.ProtoMajor == 2 && strings.Contains(r.Header.Get("Content-Type"), "application/grpc") {
grpcServer.ServeHTTP(w, r)
return
}
httpHandler.ServeHTTP(w, r)
})
}
func registerSessionForDomain(domain string, sess dtls.Session, logger logging.Logger) {
d := strings.ToLower(strings.TrimSpace(domain))
if d == "" {
@@ -1370,23 +1467,6 @@ func main() {
}
}
// 4. DTLS 서버 리스너 생성 (pion/dtls 기반)
dtlsServer, err := dtls.NewPionServer(dtls.PionServerConfig{
Addr: cfg.DTLSListen,
TLSConfig: dtlsTLSConfig,
})
if err != nil {
logger.Error("failed to start dtls server", logging.Fields{
"error": err.Error(),
})
os.Exit(1)
}
defer dtlsServer.Close()
logger.Info("dtls server listening", logging.Fields{
"addr": cfg.DTLSListen,
})
// 5. HTTP / HTTPS 서버 시작
// 프록시 타임아웃은 HOP_SERVER_PROXY_TIMEOUT_SECONDS(초 단위) 로 설정할 수 있으며,
// 기본값은 15초입니다. (ko)
@@ -1464,6 +1544,11 @@ func main() {
// 기본 HTTP → DTLS Proxy 엔트리 포인트
httpMux.Handle("/", httpHandler)
// gRPC server for client tunnels (OpenTunnel). (en)
// 클라이언트 터널(OpenTunnel)을 처리하는 gRPC 서버 인스턴스를 생성합니다. (ko)
grpcSrv := grpc.NewServer()
protocolpb.RegisterHopGateTunnelServer(grpcSrv, newGRPCTunnelServer(logger))
// HTTP: 평문 포트
httpSrv := &http.Server{
Addr: cfg.HTTPListen,
@@ -1481,9 +1566,15 @@ func main() {
}()
// HTTPS: ACME 기반 TLS 사용 (debug 모드에서도 ACME tls config 사용 가능)
// gRPC(OpenTunnel)을 위해 HTTP/2(h2)가 활성화되어 있어야 합니다. (ko)
// HTTP/2 (h2) must be enabled for gRPC (OpenTunnel) over TLS. (en)
if len(acmeTLSCfg.NextProtos) == 0 {
acmeTLSCfg.NextProtos = []string{"h2", "http/1.1"}
}
httpsSrv := &http.Server{
Addr: cfg.HTTPSListen,
Handler: httpMux,
Handler: grpcOrHTTPHandler(grpcSrv, httpMux),
TLSConfig: acmeTLSCfg,
}
go func() {
@@ -1497,89 +1588,11 @@ func main() {
}
}()
// 6. 도메인 검증기 준비 (ent + PostgreSQL 기반 실제 구현)
// Admin Plane 에서 관리하는 Domain 테이블을 사용해 (domain, client_api_key) 조합을 검증합니다.
domainValidator := admin.NewEntDomainValidator(logger, dbClient)
// 6. 도메인 검증기 준비 (향후 gRPC 터널 핸드셰이크에서 사용 예정). (ko)
// Prepare domain validator (to be used in future gRPC tunnel handshakes). (en)
_ = admin.NewEntDomainValidator(logger, dbClient)
// DTLS 핸드셰이크 단계에서는 클라이언트가 제시한 도메인의 DNS(A/AAAA)가
// HOP_ACME_EXPECT_IPS 에 설정된 IP들 중 하나 이상을 가리키는지 추가로 검증합니다. (ko)
// During DTLS handshake, additionally verify that the presented domain resolves
// (via A/AAAA) to at least one IP configured in HOP_ACME_EXPECT_IPS. (en)
// EXPECT_IPS 가 비어 있으면 DNS 기반 검증은 생략하고 DB 검증만 수행합니다. (ko)
// If EXPECT_IPS is empty, only DB-based validation is performed. (en)
expectedHandshakeIPs := parseExpectedIPsFromEnv(logger, "HOP_ACME_EXPECT_IPS")
var validator dtls.DomainValidator = &domainGateValidator{
expectedIPs: expectedHandshakeIPs,
inner: domainValidator,
logger: logger,
}
// 7. DTLS Accept 루프 + Handshake
for {
sess, err := dtlsServer.Accept()
if err != nil {
logger.Error("dtls accept failed", logging.Fields{
"error": err.Error(),
})
continue
}
// 각 세션별로 goroutine 에서 핸드셰이크 및 후속 처리를 수행합니다.
go func(s dtls.Session) {
// NOTE: 세션은 HTTP↔DTLS 터널링에 계속 사용해야 하므로 이곳에서 Close 하지 않습니다.
// 세션 종료/타임아웃 관리는 별도의 세션 매니저(TODO)에서 담당해야 합니다.
hsRes, err := dtls.PerformServerHandshake(ctx, s, validator, logger)
if err != nil {
// 핸드셰이크 실패 메트릭 기록
observability.DTLSHandshakesTotal.WithLabelValues("failure").Inc()
// PerformServerHandshake 내부에서 이미 상세 로그를 남기므로 여기서는 요약만 기록합니다.
logger.Warn("dtls handshake failed", logging.Fields{
"session_id": s.ID(),
"error": err.Error(),
})
// 핸드셰이크 실패 시 세션을 명시적으로 종료하여 invalid SNI 등 오류에서
// 연결이 열린 채로 남지 않도록 합니다.
_ = s.Close()
return
}
// Handshake 성공 메트릭 기록
observability.DTLSHandshakesTotal.WithLabelValues("success").Inc()
// Handshake 성공: 서버 측은 어떤 도메인이 연결되었는지 알 수 있습니다.
logger.Info("dtls handshake completed", logging.Fields{
"session_id": s.ID(),
"domain": hsRes.Domain,
})
// Handshake 가 완료된 세션을 도메인에 매핑해 HTTP 요청 시 사용할 수 있도록 등록합니다.
registerSessionForDomain(hsRes.Domain, s, logger)
// Handshake 가 정상적으로 끝난 이후, 실제로 해당 도메인에 대해 ACME 인증서를 확보/연장합니다.
// Debug 모드에서도 ACME 는 항상 시도하지만, 위에서 HOP_ACME_USE_STAGING=true 로 설정되어
// Staging CA 를 사용하게 됩니다.
if hsRes.Domain != "" {
go func(domain string) {
acmeLogger := logger.With(logging.Fields{
"component": "acme_post_handshake",
"domain": domain,
"debug": cfg.Debug,
})
if _, err := acme.NewLegoManagerFromEnv(context.Background(), acmeLogger, []string{domain}); err != nil {
acmeLogger.Error("failed to ensure acme certificate after dtls handshake", logging.Fields{
"error": err.Error(),
})
return
}
acmeLogger.Info("acme certificate ensured after dtls handshake", nil)
}(hsRes.Domain)
}
// TODO:
// - hsRes.Domain 과 연결된 세션을 proxy 레이어에 등록
// - HTTP 요청을 이 세션을 통해 해당 클라이언트로 라우팅
// - 세션 생명주기/타임아웃 관리 등
}(sess)
}
// DTLS 레이어 제거 이후에는 gRPC 및 HTTP/HTTPS 서버 goroutine 만 유지합니다. (ko)
// After removing the DTLS layer, only the gRPC and HTTP/HTTPS servers are kept running. (en)
select {}
}