diff --git a/.env.example b/.env.example index 4f3a8485..0f76bfda 100644 --- a/.env.example +++ b/.env.example @@ -439,12 +439,17 @@ DOCREADER_TRANSPORT=grpc # gRPC TLS 配置（可选） # GRPC_TLS_ENABLED=false -# GRPC_TLS_CERT=/path/to/server.crt -# GRPC_TLS_KEY=/path/to/server.key -# GRPC_TLS_CA=/path/to/ca.crt +# GRPC_TLS_CERT=/path/to/server.crt # server: 必填；client: mTLS 时填 +# GRPC_TLS_KEY=/path/to/server.key # 同上 +# GRPC_TLS_CA=/path/to/ca.crt # client: 校验 server 证书用；server: mTLS 时签发客户端证书的 CA +# GRPC_TLS_SERVER_NAME= # 可选；当 DOCREADER_ADDR 用 IP/与证书 CN 不一致时设为证书的 hostname +# GRPC_MTLS_REQUIRE_CLIENT_CERT=false # server 端：true 时强制要求客户端证书（mTLS） # gRPC 认证 Token（可选，客户端和服务端需配置相同的值） -# GRPC_AUTH_TOKEN=your-secret-token +# 注意：启用 GRPC_AUTH_TOKEN 时强烈建议同时启用 TLS，否则 token 会以明文随 +# 每次 RPC 在网络上传输。当 GRPC_TLS_ENABLED=true 时客户端会拒绝在不安全 +# 的连接上发送 token。 +# GRPC_AUTH_TOKEN=your-secret-token-at-least-16-bytes # Docreader 中 DOCX 解析的最大页数，默认 0（不限制） # 设为正整数（如 500）可限制超大 Word 文档的解析开销；超过页数的内容将不会继续解析 diff --git a/.env.lite.example b/.env.lite.example index 2e821cb6..cc00dd69 100644 --- a/.env.lite.example +++ b/.env.lite.example @@ -62,9 +62,14 @@ DOCREADER_TRANSPORT=grpc # gRPC TLS 配置（可选） # GRPC_TLS_ENABLED=false -# GRPC_TLS_CERT=/path/to/server.crt -# GRPC_TLS_KEY=/path/to/server.key -# GRPC_TLS_CA=/path/to/ca.crt +# GRPC_TLS_CERT=/path/to/server.crt # server: 必填；client: mTLS 时填 +# GRPC_TLS_KEY=/path/to/server.key # 同上 +# GRPC_TLS_CA=/path/to/ca.crt # client: 校验 server 证书用；server: mTLS 时签发客户端证书的 CA +# GRPC_TLS_SERVER_NAME= # 可选；当 DOCREADER_ADDR 用 IP/与证书 CN 不一致时设为证书的 hostname +# GRPC_MTLS_REQUIRE_CLIENT_CERT=false # server 端：true 时强制要求客户端证书（mTLS） # gRPC 认证 Token（可选，客户端和服务端需配置相同的值） -# GRPC_AUTH_TOKEN=your-secret-token \ No newline at end of file +# 注意：启用 GRPC_AUTH_TOKEN 时强烈建议同时启用 TLS，否则 token 会以明文随 +# 每次 RPC 在网络上传输。当 GRPC_TLS_ENABLED=true 时客户端会拒绝在不安全 +# 的连接上发送 token。 +# GRPC_AUTH_TOKEN=your-secret-token-at-least-16-bytes diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 5efd4e2d..972d7ef8 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -203,6 +203,7 @@ services: - GRPC_TLS_CERT=${GRPC_TLS_CERT:-} - GRPC_TLS_KEY=${GRPC_TLS_KEY:-} - GRPC_TLS_CA=${GRPC_TLS_CA:-} + - GRPC_TLS_SERVER_NAME=${GRPC_TLS_SERVER_NAME:-} - GRPC_AUTH_TOKEN=${GRPC_AUTH_TOKEN:-} healthcheck: test: ["CMD", "grpc_health_probe", "-addr=localhost:50051"] diff --git a/docker-compose.yml b/docker-compose.yml index 99422de3..745e481e 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -102,6 +102,13 @@ services: - MILVUS_METRIC_TYPE=${MILVUS_METRIC_TYPE:-IP} - DOCREADER_ADDR=${DOCREADER_ADDR:-docreader:50051} - DOCREADER_TRANSPORT=${DOCREADER_TRANSPORT:-grpc} + # docreader gRPC TLS / 认证（客户端侧） + - GRPC_TLS_ENABLED=${GRPC_TLS_ENABLED:-false} + - GRPC_TLS_CERT=${GRPC_TLS_CERT:-} + - GRPC_TLS_KEY=${GRPC_TLS_KEY:-} + - GRPC_TLS_CA=${GRPC_TLS_CA:-} + - GRPC_TLS_SERVER_NAME=${GRPC_TLS_SERVER_NAME:-} + - GRPC_AUTH_TOKEN=${GRPC_AUTH_TOKEN:-} - WEAVIATE_HOST=${WEAVIATE_HOST:-weaviate:8080} - WEAVIATE_GRPC_ADDRESS=${WEAVIATE_GRPC_ADDRESS:-weaviate:50051} - WEAVIATE_SCHEME=${WEAVIATE_SCHEME:-http} diff --git a/docreader/auth.py b/docreader/auth.py index c6d440a7..6667e27b 100644 --- a/docreader/auth.py +++ b/docreader/auth.py @@ -3,14 +3,21 @@ 环境变量配置： TLS 相关： GRPC_TLS_ENABLED: 是否启用 TLS（true/false），默认 false - GRPC_TLS_CERT: TLS 证书文件路径 - GRPC_TLS_KEY: TLS 私钥文件路径 - GRPC_TLS_CA: CA 证书路径（可选，用于 mTLS 双向认证） + GRPC_TLS_CERT: TLS 证书文件路径（GRPC_TLS_ENABLED=true 时必填） + GRPC_TLS_KEY: TLS 私钥文件路径（GRPC_TLS_ENABLED=true 时必填） + GRPC_TLS_CA: CA 证书路径 + GRPC_MTLS_REQUIRE_CLIENT_CERT: 设为 true 时启用 mTLS，要求客户端 + 出示由 GRPC_TLS_CA 签发的证书。未设置时默认按 GRPC_TLS_CA 是否 + 存在自动判断（保留向后兼容）。 认证相关： GRPC_AUTH_TOKEN: 认证 Token，如果设置则启用认证 + +注意：当 GRPC_TLS_ENABLED=true 但任何 TLS 配置项缺失或加载失败时， +本模块会抛出异常以触发 fail-fast，避免静默降级到明文。 """ +import hmac import logging import os from typing import Optional @@ -20,9 +27,24 @@ import grpc logger = logging.getLogger(__name__) +class TLSConfigError(RuntimeError): + """TLS 配置错误，用于 fail-fast。""" + + +def _env_bool(name: str, default: bool = False) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return raw.strip().lower() in ("true", "1", "yes", "on") + + def load_tls_credentials() -> Optional[grpc.ServerCredentials]: - tls_enabled = os.getenv("GRPC_TLS_ENABLED", "false").lower() == "true" - if not tls_enabled: + """构建 server 端 TLS 凭据。 + + GRPC_TLS_ENABLED=false 时返回 None；为 true 时如果配置无效会抛出 + TLSConfigError，由调用方决定是否终止启动。 + """ + if not _env_bool("GRPC_TLS_ENABLED", False): logger.info("TLS disabled (GRPC_TLS_ENABLED is not 'true')") return None @@ -30,38 +52,105 @@ def load_tls_credentials() -> Optional[grpc.ServerCredentials]: key_path = os.getenv("GRPC_TLS_KEY") if not cert_path or not key_path: - logger.warning( - "TLS enabled but certificate not configured (GRPC_TLS_CERT or GRPC_TLS_KEY not set)" + raise TLSConfigError( + "GRPC_TLS_ENABLED=true but GRPC_TLS_CERT/GRPC_TLS_KEY not set; " + "refusing to start in plaintext mode" ) - return None try: with open(cert_path, "rb") as f: cert_chain = f.read() with open(key_path, "rb") as f: private_key = f.read() + except OSError as e: + raise TLSConfigError(f"failed to read TLS cert/key: {e}") from e - ca_path = os.getenv("GRPC_TLS_CA") - if ca_path: + ca_path = os.getenv("GRPC_TLS_CA") + require_client_auth = _env_bool( + "GRPC_MTLS_REQUIRE_CLIENT_CERT", + default=bool(ca_path), + ) + + if require_client_auth and not ca_path: + raise TLSConfigError( + "GRPC_MTLS_REQUIRE_CLIENT_CERT=true requires GRPC_TLS_CA to be set" + ) + + if ca_path: + try: with open(ca_path, "rb") as f: ca_cert = f.read() - credentials = grpc.ssl_server_credentials( - [(private_key, cert_chain)], - root_certificates=ca_cert, - require_client_auth=True, - ) + except OSError as e: + raise TLSConfigError(f"failed to read CA cert: {e}") from e + credentials = grpc.ssl_server_credentials( + [(private_key, cert_chain)], + root_certificates=ca_cert, + require_client_auth=require_client_auth, + ) + if require_client_auth: logger.info("TLS enabled with mTLS (mutual authentication)") else: - credentials = grpc.ssl_server_credentials([(private_key, cert_chain)]) - logger.info("TLS enabled") + logger.info("TLS enabled with CA configured (client auth optional)") + else: + credentials = grpc.ssl_server_credentials([(private_key, cert_chain)]) + logger.info("TLS enabled (1-way)") - return credentials - except FileNotFoundError as e: - logger.error(f"TLS certificate file not found: {e}") - return None - except Exception as e: - logger.error(f"Failed to load TLS credentials: {e}") - return None + return credentials + + +# gRPC 健康检查标准服务路径，需在鉴权前放行，便于 K8s/Docker 探活。 +_HEALTH_METHODS = frozenset( + { + "/grpc.health.v1.Health/Check", + "/grpc.health.v1.Health/Watch", + } +) + + +def _make_abort_handler(original: Optional[grpc.RpcMethodHandler]) -> grpc.RpcMethodHandler: + """为给定的原 handler 构造一个匹配 RPC kind 的鉴权失败 handler。 + + 如果原 handler 不可用（理论上不会发生，但作为兜底），返回 unary_unary。 + 直接返回 None 而不是 set_code 也可，但显式调用 abort 能确保框架按 + UNAUTHENTICATED 收尾，且匹配 kind 防止 grpc 触发 INTERNAL。 + """ + def _abort(_request, context): + context.abort( + grpc.StatusCode.UNAUTHENTICATED, + "Invalid or missing authentication token", + ) + + def _abort_stream(_request, context): + context.abort( + grpc.StatusCode.UNAUTHENTICATED, + "Invalid or missing authentication token", + ) + return + yield # pragma: no cover - make this a generator + + if original is None or original.unary_unary is not None: + return grpc.unary_unary_rpc_method_handler( + _abort, + request_deserializer=getattr(original, "request_deserializer", None), + response_serializer=getattr(original, "response_serializer", None), + ) + if original.unary_stream is not None: + return grpc.unary_stream_rpc_method_handler( + _abort_stream, + request_deserializer=original.request_deserializer, + response_serializer=original.response_serializer, + ) + if original.stream_unary is not None: + return grpc.stream_unary_rpc_method_handler( + _abort, + request_deserializer=original.request_deserializer, + response_serializer=original.response_serializer, + ) + return grpc.stream_stream_rpc_method_handler( + _abort_stream, + request_deserializer=original.request_deserializer, + response_serializer=original.response_serializer, + ) class AuthInterceptor(grpc.ServerInterceptor): @@ -75,9 +164,14 @@ class AuthInterceptor(grpc.ServerInterceptor): - value: "Bearer " 或直接 "" """ - def __init__(self): - self.auth_token = os.getenv("GRPC_AUTH_TOKEN") + def __init__(self) -> None: + token = os.getenv("GRPC_AUTH_TOKEN") or "" + self.auth_token: Optional[bytes] = token.encode("utf-8") if token else None if self.auth_token: + if len(self.auth_token) < 16: + logger.warning( + "GRPC_AUTH_TOKEN is shorter than 16 bytes; consider a stronger token" + ) logger.info("Token authentication enabled") else: logger.warning("Token authentication disabled (GRPC_AUTH_TOKEN not set)") @@ -87,25 +181,18 @@ class AuthInterceptor(grpc.ServerInterceptor): return continuation(handler_call_details) method = handler_call_details.method - if method.endswith("/Check") or method.endswith("/Watch"): + if method in _HEALTH_METHODS: return continuation(handler_call_details) metadata = dict(handler_call_details.invocation_metadata or []) - token = metadata.get("authorization", "") + raw = metadata.get("authorization", "") or "" + if raw.startswith("Bearer "): + raw = raw[7:] + token_bytes = raw.encode("utf-8") - if token.startswith("Bearer "): - token = token[7:] - - if token != self.auth_token: - logger.warning(f"Authentication failed for method: {method}") - return self._unauthenticated_handler() + if not hmac.compare_digest(token_bytes, self.auth_token): + logger.warning("Authentication failed for method: %s", method) + original = continuation(handler_call_details) + return _make_abort_handler(original) return continuation(handler_call_details) - - def _unauthenticated_handler(self): - def handler(request, context): - context.set_code(grpc.StatusCode.UNAUTHENTICATED) - context.set_details("Invalid or missing authentication token") - return None - - return grpc.unary_unary_rpc_method_handler(handler) diff --git a/docreader/client/auth.go b/docreader/client/auth.go index e05e4a7a..aa4d79ea 100644 --- a/docreader/client/auth.go +++ b/docreader/client/auth.go @@ -1,3 +1,7 @@ +// Package client provides a docreader gRPC client and the shared TLS / token +// authentication helpers used by both the standalone Go SDK in this package +// and the internal docparser wrapper. Keep all auth/TLS construction here so +// the two call sites cannot drift on security defaults. package client import ( @@ -12,25 +16,36 @@ import ( "google.golang.org/grpc/credentials/insecure" ) +// AuthConfig holds the docreader gRPC client TLS / token configuration. type AuthConfig struct { TLSEnabled bool CertFile string KeyFile string CAFile string + // ServerName overrides the SNI / certificate-host check on the client. + // When empty, the address passed to Dial is used by Go's TLS stack. + ServerName string AuthToken string } +// LoadAuthConfigFromEnv reads docreader gRPC auth knobs from the process +// environment. The caller must pass the result to BuildDialOptions to apply +// them to a gRPC connection. func LoadAuthConfigFromEnv() *AuthConfig { return &AuthConfig{ TLSEnabled: os.Getenv("GRPC_TLS_ENABLED") == "true", CertFile: os.Getenv("GRPC_TLS_CERT"), KeyFile: os.Getenv("GRPC_TLS_KEY"), CAFile: os.Getenv("GRPC_TLS_CA"), + ServerName: os.Getenv("GRPC_TLS_SERVER_NAME"), AuthToken: os.Getenv("GRPC_AUTH_TOKEN"), } } +// BuildDialOptions returns the gRPC DialOptions that apply the configured +// transport credentials and per-RPC token. Callers should append their own +// per-call options (load balancer, message size, etc.). func (c *AuthConfig) BuildDialOptions(maxMsgSize int) ([]grpc.DialOption, error) { opts := []grpc.DialOption{ grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`), @@ -52,8 +67,14 @@ func (c *AuthConfig) BuildDialOptions(maxMsgSize int) ([]grpc.DialOption, error) } if c.AuthToken != "" { - opts = append(opts, grpc.WithPerRPCCredentials(&tokenAuth{token: c.AuthToken})) - Logger.Printf("INFO: Token authentication enabled for gRPC client") + // Only allow per-RPC tokens to ride a secured channel. This mirrors + // gRPC's own oauth2 credentials behaviour and prevents the bearer + // token from leaking on plaintext connections. + opts = append(opts, grpc.WithPerRPCCredentials(&tokenAuth{ + token: c.AuthToken, + requireTLSGuard: c.TLSEnabled, + })) + Logger.Printf("INFO: Token authentication enabled for gRPC client (TLS=%v)", c.TLSEnabled) } return opts, nil @@ -62,6 +83,7 @@ func (c *AuthConfig) BuildDialOptions(maxMsgSize int) ([]grpc.DialOption, error) func (c *AuthConfig) buildTLSCredentials() (credentials.TransportCredentials, error) { tlsConfig := &tls.Config{ MinVersion: tls.VersionTLS12, + ServerName: c.ServerName, } if c.CAFile != "" { @@ -76,13 +98,18 @@ func (c *AuthConfig) buildTLSCredentials() (credentials.TransportCredentials, er tlsConfig.RootCAs = certPool } - if c.CertFile != "" && c.KeyFile != "" { + switch { + case c.CertFile != "" && c.KeyFile != "": cert, err := tls.LoadX509KeyPair(c.CertFile, c.KeyFile) if err != nil { return nil, fmt.Errorf("failed to load client certificate: %w", err) } tlsConfig.Certificates = []tls.Certificate{cert} Logger.Printf("INFO: mTLS enabled (client certificate loaded)") + case c.CertFile != "" || c.KeyFile != "": + return nil, fmt.Errorf( + "GRPC_TLS_CERT and GRPC_TLS_KEY must be set together for mTLS", + ) } return credentials.NewTLS(tlsConfig), nil @@ -90,6 +117,10 @@ func (c *AuthConfig) buildTLSCredentials() (credentials.TransportCredentials, er type tokenAuth struct { token string + // requireTLSGuard mirrors AuthConfig.TLSEnabled; we expose it via + // RequireTransportSecurity so gRPC will refuse to send the bearer token + // over an insecure connection when the operator has enabled TLS. + requireTLSGuard bool } func (t *tokenAuth) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) { @@ -99,5 +130,5 @@ func (t *tokenAuth) GetRequestMetadata(ctx context.Context, uri ...string) (map[ } func (t *tokenAuth) RequireTransportSecurity() bool { - return false + return t.requireTLSGuard } diff --git a/docreader/main.py b/docreader/main.py index 9caa1931..6f915a50 100644 --- a/docreader/main.py +++ b/docreader/main.py @@ -11,7 +11,7 @@ import grpc from grpc_health.v1 import health_pb2_grpc from grpc_health.v1.health import HealthServicer -from docreader.auth import AuthInterceptor, load_tls_credentials +from docreader.auth import AuthInterceptor, TLSConfigError, load_tls_credentials from docreader import config from docreader.config import CONFIG from docreader.parser import Parser @@ -212,7 +212,12 @@ def main(): health_servicer = HealthServicer() health_pb2_grpc.add_HealthServicer_to_server(health_servicer, server) - tls_credentials = load_tls_credentials() + try: + tls_credentials = load_tls_credentials() + except TLSConfigError as e: + logger.error("Refusing to start: %s", e) + sys.exit(1) + if tls_credentials: server.add_secure_port(f"[::]:{CONFIG.grpc_port}", tls_credentials) logger.info("Server starting on port %d with TLS", CONFIG.grpc_port) diff --git a/internal/infrastructure/docparser/grpc_parser.go b/internal/infrastructure/docparser/grpc_parser.go index f28caa24..61d2d15b 100644 --- a/internal/infrastructure/docparser/grpc_parser.go +++ b/internal/infrastructure/docparser/grpc_parser.go @@ -2,20 +2,17 @@ package docparser import ( "context" - "crypto/tls" - "crypto/x509" "fmt" "os" "strconv" "sync" "time" + docclient "github.com/Tencent/WeKnora/docreader/client" "github.com/Tencent/WeKnora/docreader/proto" "github.com/Tencent/WeKnora/internal/logger" "github.com/Tencent/WeKnora/internal/types" "google.golang.org/grpc" - "google.golang.org/grpc/credentials" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/resolver" ) @@ -47,30 +44,19 @@ func NewGRPCDocumentReader(addr string) (*GRPCDocumentReader, error) { } func (p *GRPCDocumentReader) connect(addr string) error { - - maxMsgSize := getMaxMessageSize() - opts := []grpc.DialOption{ - grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`), - grpc.WithDefaultCallOptions( - grpc.MaxCallRecvMsgSize(maxMsgSize), - grpc.MaxCallSendMsgSize(maxMsgSize), - ), + authConfig := docclient.LoadAuthConfigFromEnv() + opts, err := authConfig.BuildDialOptions(getMaxMessageSize()) + if err != nil { + return fmt.Errorf("failed to build docreader dial options: %w", err) } - - if os.Getenv("GRPC_TLS_ENABLED") == "true" { - creds, err := buildTLSCredentials() - if err != nil { - return fmt.Errorf("failed to build TLS credentials: %w", err) - } - opts = append(opts, grpc.WithTransportCredentials(creds)) + if authConfig.TLSEnabled { logger.Infof(context.Background(), "TLS enabled for docreader gRPC client") - } else { - opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials())) } - - if authToken := os.Getenv("GRPC_AUTH_TOKEN"); authToken != "" { - opts = append(opts, grpc.WithPerRPCCredentials(&tokenAuth{token: authToken})) - logger.Infof(context.Background(), "Token authentication enabled for docreader gRPC client") + if authConfig.AuthToken != "" { + logger.Infof(context.Background(), + "Token authentication enabled for docreader gRPC client (TLS=%v)", + authConfig.TLSEnabled, + ) } resolver.SetDefaultScheme("dns") @@ -192,48 +178,3 @@ func fromProtoReadResponse(resp *proto.ReadResponse) *types.ReadResult { return result } - -func buildTLSCredentials() (credentials.TransportCredentials, error) { - tlsConfig := &tls.Config{ - MinVersion: tls.VersionTLS12, - } - - if caFile := os.Getenv("GRPC_TLS_CA"); caFile != "" { - caCert, err := os.ReadFile(caFile) - if err != nil { - return nil, fmt.Errorf("failed to read CA certificate: %w", err) - } - certPool := x509.NewCertPool() - if !certPool.AppendCertsFromPEM(caCert) { - return nil, fmt.Errorf("failed to parse CA certificate") - } - tlsConfig.RootCAs = certPool - } - - if certFile := os.Getenv("GRPC_TLS_CERT"); certFile != "" { - if keyFile := os.Getenv("GRPC_TLS_KEY"); keyFile != "" { - cert, err := tls.LoadX509KeyPair(certFile, keyFile) - if err != nil { - return nil, fmt.Errorf("failed to load client certificate: %w", err) - } - tlsConfig.Certificates = []tls.Certificate{cert} - logger.Infof(context.Background(), "mTLS enabled (client certificate loaded)") - } - } - - return credentials.NewTLS(tlsConfig), nil -} - -type tokenAuth struct { - token string -} - -func (t *tokenAuth) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) { - return map[string]string{ - "authorization": "Bearer " + t.token, - }, nil -} - -func (t *tokenAuth) RequireTransportSecurity() bool { - return false -}