gRPC PERMISSION_DENIED — Auth and Authorization Issues
Belirtiler
- Client error: `StatusCode.PERMISSION_DENIED: Access denied` or `insufficient permissions`
- Status code 7 returned immediately without server-side processing
- Interceptor logs show `authorization` metadata key missing from request
- Works with admin credentials but fails with regular user token
- RBAC audit log shows the principal lacks the required role for the RPC method
- Status code 7 returned immediately without server-side processing
- Interceptor logs show `authorization` metadata key missing from request
- Works with admin credentials but fails with regular user token
- RBAC audit log shows the principal lacks the required role for the RPC method
Kök Nedenler
- Bearer token not attached to gRPC metadata — client forgot to add Authorization header
- JWT token expired — access token lifetime elapsed since last refresh
- Token has insufficient OAuth scopes for the called gRPC method
- Server-side RBAC policy does not grant the caller's role access to this method
- mTLS client certificate not provided or not trusted by the server CA
Tanı
**Step 1 — Check if metadata is being sent**
```python
import grpc
# Add interceptor to log outgoing metadata
class MetadataLogInterceptor(grpc.UnaryUnaryClientInterceptor):
def intercept_unary_unary(self, continuation, client_call_details, request):
print('Metadata:', client_call_details.metadata)
return continuation(client_call_details, request)
```
**Step 2 — Inspect the JWT token**
```bash
# Decode JWT without verifying signature (for inspection only)
TOKEN='eyJhbGciOiJSUzI1NiJ9...'
echo $TOKEN | cut -d. -f2 | base64 -d 2>/dev/null | python3 -m json.tool
# Check: exp (expiry), scope, sub, aud
```
**Step 3 — Call with grpcurl passing the token**
```bash
grpcurl -plaintext \
-H 'authorization: Bearer eyJhbGci...' \
-d '{"resource_id": "123"}' \
localhost:50051 mypackage.MyService/GetResource
# If this works but your code doesn't, the client is not attaching the token
```
**Step 4 — Check server interceptor logs**
```python
# Server-side interceptor to log auth decisions
def intercept_service(self, handler_call_details):
metadata = dict(handler_call_details.invocation_metadata)
auth_header = metadata.get('authorization', 'MISSING')
print(f'Auth header: {auth_header[:30]}...')
return self.continuation(handler_call_details)
```
**Step 5 — Review RBAC policy**
```yaml
# Example OPA policy check
opa eval -d policy.rego \
-I '{"method": "/mypackage.MyService/GetResource", "token": {"sub": "user123", "roles": ["viewer"]}}' \
'data.grpc.authz.allow'
```
```python
import grpc
# Add interceptor to log outgoing metadata
class MetadataLogInterceptor(grpc.UnaryUnaryClientInterceptor):
def intercept_unary_unary(self, continuation, client_call_details, request):
print('Metadata:', client_call_details.metadata)
return continuation(client_call_details, request)
```
**Step 2 — Inspect the JWT token**
```bash
# Decode JWT without verifying signature (for inspection only)
TOKEN='eyJhbGciOiJSUzI1NiJ9...'
echo $TOKEN | cut -d. -f2 | base64 -d 2>/dev/null | python3 -m json.tool
# Check: exp (expiry), scope, sub, aud
```
**Step 3 — Call with grpcurl passing the token**
```bash
grpcurl -plaintext \
-H 'authorization: Bearer eyJhbGci...' \
-d '{"resource_id": "123"}' \
localhost:50051 mypackage.MyService/GetResource
# If this works but your code doesn't, the client is not attaching the token
```
**Step 4 — Check server interceptor logs**
```python
# Server-side interceptor to log auth decisions
def intercept_service(self, handler_call_details):
metadata = dict(handler_call_details.invocation_metadata)
auth_header = metadata.get('authorization', 'MISSING')
print(f'Auth header: {auth_header[:30]}...')
return self.continuation(handler_call_details)
```
**Step 5 — Review RBAC policy**
```yaml
# Example OPA policy check
opa eval -d policy.rego \
-I '{"method": "/mypackage.MyService/GetResource", "token": {"sub": "user123", "roles": ["viewer"]}}' \
'data.grpc.authz.allow'
```
Çözüm
**Fix 1 — Attach token to gRPC metadata**
```python
import grpc
def make_call_credentials(token: str) -> grpc.CallCredentials:
def metadata_callback(context, callback):
callback([('authorization', f'Bearer {token}')], None)
return grpc.metadata_call_credentials(metadata_callback)
# Attach per-call
creds = make_call_credentials(my_token)
response = stub.MyMethod(request, credentials=creds)
```
**Fix 2 — Refresh expired token before each call**
```python
import time
from functools import lru_cache
class TokenManager:
def __init__(self):
self._token = None
self._expires_at = 0
def get_token(self) -> str:
if time.time() > self._expires_at - 60: # 60s buffer
self._token, self._expires_at = self._fetch_new_token()
return self._token
```
**Fix 3 — Request correct OAuth scopes**
```python
# Google Cloud example
from google.oauth2 import service_account
import google.auth.transport.grpc
credentials = service_account.Credentials.from_service_account_file(
'service-account.json',
scopes=['https://www.googleapis.com/auth/cloud-platform'],
)
```
**Fix 4 — Update RBAC role binding**
```yaml
# Kubernetes RBAC for gRPC via Envoy/Istio AuthorizationPolicy
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
name: my-grpc-policy
spec:
rules:
- from:
- source:
principals: ['cluster.local/ns/default/sa/my-service-account']
to:
- operation:
paths: ['/mypackage.MyService/*']
```
```python
import grpc
def make_call_credentials(token: str) -> grpc.CallCredentials:
def metadata_callback(context, callback):
callback([('authorization', f'Bearer {token}')], None)
return grpc.metadata_call_credentials(metadata_callback)
# Attach per-call
creds = make_call_credentials(my_token)
response = stub.MyMethod(request, credentials=creds)
```
**Fix 2 — Refresh expired token before each call**
```python
import time
from functools import lru_cache
class TokenManager:
def __init__(self):
self._token = None
self._expires_at = 0
def get_token(self) -> str:
if time.time() > self._expires_at - 60: # 60s buffer
self._token, self._expires_at = self._fetch_new_token()
return self._token
```
**Fix 3 — Request correct OAuth scopes**
```python
# Google Cloud example
from google.oauth2 import service_account
import google.auth.transport.grpc
credentials = service_account.Credentials.from_service_account_file(
'service-account.json',
scopes=['https://www.googleapis.com/auth/cloud-platform'],
)
```
**Fix 4 — Update RBAC role binding**
```yaml
# Kubernetes RBAC for gRPC via Envoy/Istio AuthorizationPolicy
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
name: my-grpc-policy
spec:
rules:
- from:
- source:
principals: ['cluster.local/ns/default/sa/my-service-account']
to:
- operation:
paths: ['/mypackage.MyService/*']
```
Önleme
- Use a token manager class that proactively refreshes tokens 60 seconds before expiry
- Add a client-side interceptor that attaches auth metadata to every RPC automatically
- Write integration tests that verify auth header presence using a mock server interceptor
- Implement fine-grained RBAC at the method level and test each role in CI
- Use mTLS for service-to-service gRPC to eliminate bearer token management entirely
- Add a client-side interceptor that attaches auth metadata to every RPC automatically
- Write integration tests that verify auth header presence using a mock server interceptor
- Implement fine-grained RBAC at the method level and test each role in CI
- Use mTLS for service-to-service gRPC to eliminate bearer token management entirely