UNAUTHENTICATED — Missing or Invalid gRPC Metadata
症状
- gRPC calls fail with `StatusCode.UNAUTHENTICATED` and details: "Missing or invalid authorization metadata"
- The RPC works from one service but not another — auth metadata is injected in one but not the other
- Token expired and the client has no refresh logic — calls fail silently until restart
- Metadata key is wrong: gRPC metadata keys must be lowercase ("authorization", not "Authorization")
- Service account key file missing or unreadable in the production environment
- The RPC works from one service but not another — auth metadata is injected in one but not the other
- Token expired and the client has no refresh logic — calls fail silently until restart
- Metadata key is wrong: gRPC metadata keys must be lowercase ("authorization", not "Authorization")
- Service account key file missing or unreadable in the production environment
根本原因
- Bearer token (JWT) not attached to the gRPC call metadata
- Token expired between client startup and the actual RPC — no refresh interceptor configured
- Metadata key casing is wrong: gRPC requires all metadata keys to be lowercase
- Client interceptor/middleware that injects auth token is missing or misconfigured in production
- Service account credentials file path is hardcoded to a local path that doesn't exist on the server
诊断
**Step 1 — Inspect metadata with grpcurl**
```bash
# List available methods and test with auth header
grpcurl \
-H 'authorization: Bearer YOUR_TOKEN' \
-d '{}' \
api.example.com:443 \
mypackage.MyService/GetItem
# Without the header — expect UNAUTHENTICATED
grpcurl -d '{}' api.example.com:443 mypackage.MyService/GetItem
```
**Step 2 — Verify the JWT is not expired**
```python
import jwt, time
payload = jwt.decode(token, options={'verify_signature': False})
print(f'Expires at: {payload["exp"]}, now: {int(time.time())}')
# If exp < now, the token is expired
```
Use the JWT Decoder tool above to inspect the token without code.
**Step 3 — Check metadata key casing**
```python
# WRONG — HTTP-style capitalization (silently ignored by many servers)
metadata = [('Authorization', f'Bearer {token}')]
# CORRECT — gRPC metadata keys are always lowercase
metadata = [('authorization', f'Bearer {token}')]
```
**Step 4 — Log outgoing metadata in development**
```python
import grpc
import os
os.environ['GRPC_VERBOSITY'] = 'debug'
os.environ['GRPC_TRACE'] = 'metadata'
# Restart your app — gRPC will log all outgoing and incoming metadata
```
```bash
# List available methods and test with auth header
grpcurl \
-H 'authorization: Bearer YOUR_TOKEN' \
-d '{}' \
api.example.com:443 \
mypackage.MyService/GetItem
# Without the header — expect UNAUTHENTICATED
grpcurl -d '{}' api.example.com:443 mypackage.MyService/GetItem
```
**Step 2 — Verify the JWT is not expired**
```python
import jwt, time
payload = jwt.decode(token, options={'verify_signature': False})
print(f'Expires at: {payload["exp"]}, now: {int(time.time())}')
# If exp < now, the token is expired
```
Use the JWT Decoder tool above to inspect the token without code.
**Step 3 — Check metadata key casing**
```python
# WRONG — HTTP-style capitalization (silently ignored by many servers)
metadata = [('Authorization', f'Bearer {token}')]
# CORRECT — gRPC metadata keys are always lowercase
metadata = [('authorization', f'Bearer {token}')]
```
**Step 4 — Log outgoing metadata in development**
```python
import grpc
import os
os.environ['GRPC_VERBOSITY'] = 'debug'
os.environ['GRPC_TRACE'] = 'metadata'
# Restart your app — gRPC will log all outgoing and incoming metadata
```
解决
**Fix 1 — Attach the token per-call**
```python
import grpc
token = get_access_token()
metadata = [('authorization', f'Bearer {token}')]
response = stub.GetItem(request, metadata=metadata)
```
**Fix 2 — Use a client interceptor for automatic injection**
```python
import grpc
from grpc import UnaryUnaryClientInterceptor, ClientCallDetails
class AuthInterceptor(UnaryUnaryClientInterceptor):
def __init__(self, get_token):
self.get_token = get_token
def intercept_unary_unary(self, continuation, client_call_details, request):
token = self.get_token()
metadata = list(client_call_details.metadata or [])
metadata.append(('authorization', f'Bearer {token}'))
new_details = ClientCallDetails()
new_details.method = client_call_details.method
new_details.timeout = client_call_details.timeout
new_details.metadata = metadata
return continuation(new_details, request)
# Apply to channel
channel = grpc.intercept_channel(base_channel, AuthInterceptor(get_token_func))
```
**Fix 3 — Use gRPC's built-in CallCredentials**
```python
def token_fetcher(context, callback):
token = refresh_token_if_needed() # handles expiry automatically
callback([('authorization', f'Bearer {token}')], None)
call_credentials = grpc.metadata_call_credentials(token_fetcher)
channel_credentials = grpc.ssl_channel_credentials()
combined = grpc.composite_channel_credentials(channel_credentials, call_credentials)
channel = grpc.secure_channel('api.example.com:443', combined)
```
```python
import grpc
token = get_access_token()
metadata = [('authorization', f'Bearer {token}')]
response = stub.GetItem(request, metadata=metadata)
```
**Fix 2 — Use a client interceptor for automatic injection**
```python
import grpc
from grpc import UnaryUnaryClientInterceptor, ClientCallDetails
class AuthInterceptor(UnaryUnaryClientInterceptor):
def __init__(self, get_token):
self.get_token = get_token
def intercept_unary_unary(self, continuation, client_call_details, request):
token = self.get_token()
metadata = list(client_call_details.metadata or [])
metadata.append(('authorization', f'Bearer {token}'))
new_details = ClientCallDetails()
new_details.method = client_call_details.method
new_details.timeout = client_call_details.timeout
new_details.metadata = metadata
return continuation(new_details, request)
# Apply to channel
channel = grpc.intercept_channel(base_channel, AuthInterceptor(get_token_func))
```
**Fix 3 — Use gRPC's built-in CallCredentials**
```python
def token_fetcher(context, callback):
token = refresh_token_if_needed() # handles expiry automatically
callback([('authorization', f'Bearer {token}')], None)
call_credentials = grpc.metadata_call_credentials(token_fetcher)
channel_credentials = grpc.ssl_channel_credentials()
combined = grpc.composite_channel_credentials(channel_credentials, call_credentials)
channel = grpc.secure_channel('api.example.com:443', combined)
```
预防
- Always use a client interceptor or CallCredentials for token injection — per-call metadata attachment is error-prone and easy to forget
- Implement token refresh logic inside the credential fetcher, not as a one-time setup at startup
- Use lowercase for all gRPC metadata keys — create a constant (`AUTH_HEADER = 'authorization'`) and use it everywhere
- Write an integration test that calls each protected RPC without a token and asserts UNAUTHENTICATED — confirms auth is enforced
- Store service account credentials in a secrets manager and inject the path via environment variable, never hardcode the file path
- Implement token refresh logic inside the credential fetcher, not as a one-time setup at startup
- Use lowercase for all gRPC metadata keys — create a constant (`AUTH_HEADER = 'authorization'`) and use it everywhere
- Write an integration test that calls each protected RPC without a token and asserts UNAUTHENTICATED — confirms auth is enforced
- Store service account credentials in a secrets manager and inject the path via environment variable, never hardcode the file path