Crossplane Security Testing: Provider Credential Exposure, Composition Injection, and RBAC Privilege Escalation

Provider Security and Credential Audit

# Crossplane providers (AWS, GCP, Azure) run as Deployments in crossplane-system
# They hold cloud credentials used to manage infrastructure

# List installed providers and their status
kubectl get providers -o json | python3 -c "
import json,sys
providers = json.load(sys.stdin)['items']
for p in providers:
    meta = p['metadata']
    status = p.get('status',{})
    conditions = status.get('conditions',[])
    healthy = any(c.get('type')=='Healthy' and c.get('status')=='True' for c in conditions)
    print(f\"Provider: {meta['name']} healthy: {healthy}\")
    print(f\"  Package: {p.get('spec',{}).get('package','?')}\")
"

# Check provider pods for credential environment variables
kubectl get pods -n crossplane-system -o json | python3 -c "
import json,sys
pods = json.load(sys.stdin)['items']
for p in pods:
    if 'provider' in p['metadata']['name'].lower():
        for c in p['spec'].get('containers',[]):
            for env in c.get('env',[]):
                name = env.get('name','')
                if any(k in name.upper() for k in ['AWS','GCP','AZURE','KEY','SECRET','TOKEN','CRED']):
                    val_from = env.get('valueFrom',{}).get('secretKeyRef',{})
                    print(f\"{p['metadata']['name']}/{c['name']}: {name} from secret: {val_from.get('name','direct')}\")
"

# Check ProviderConfigs — they reference the credential secrets
kubectl get providerconfigs --all-namespaces -o json | python3 -c "
import json,sys
pcs = json.load(sys.stdin)['items']
for pc in pcs:
    meta = pc['metadata']
    spec = pc.get('spec',{})
    creds = spec.get('credentials',{})
    secret_ref = creds.get('secretRef',{})
    print(f\"ProviderConfig: {meta.get('namespace','cluster')}/{meta['name']}\")
    print(f\"  Source: {creds.get('source','?')}\")
    print(f\"  Secret: {secret_ref.get('namespace','?')}/{secret_ref.get('name','?')} key: {secret_ref.get('key','?')}\")
"

ProviderConfig Manipulation

# ProviderConfig points to a credential secret — if you can create/modify ProviderConfigs,
# you can point managed resources to use credentials you control

# Check who can create ProviderConfigs
kubectl auth can-i create providerconfigs \
  --as=system:serviceaccount:default:default
kubectl auth can-i create providerconfigs.aws.crossplane.io \
  --as=system:serviceaccount:app-team:app-service-account

# If allowed: create a ProviderConfig pointing to stolen credentials
# Or: create a ProviderConfig pointing to credentials in your namespace
kubectl apply -f - </dev/null | python3 -c "
import json,sys
try:
    resources = json.load(sys.stdin)['items']
    for r in resources:
        pc_ref = r.get('spec',{}).get('providerConfigRef',{})
        meta = r['metadata']
        print(f\"{meta['name']}: providerConfig={pc_ref.get('name','default')}\")
except: print('No managed resources or auth required')
" | head -20

Managed Resource RBAC Testing

# Managed resources represent cloud infrastructure
# Creating managed resources = creating cloud resources (S3 buckets, VPCs, VMs, etc.)
# Test who can create managed resources

# Check managed resource CRDs available
kubectl api-resources | grep -E "crossplane|upbound|aws|gcp|azure" | head -20

# Test if a low-privilege user can create cloud resources
# Creating an S3 bucket via Crossplane:
kubectl auth can-i create buckets.s3.aws.crossplane.io \
  --as=system:serviceaccount:developer:developer-sa

# Test: create an S3 bucket from developer namespace
kubectl apply -f - <

# Compositions define how composite resources are transformed into managed resources
# User-controlled fields in Claims flow into Composition pipelines
# Injection: can a user set fields in a Claim that affect the cloud resource beyond intent?

# Check Composition inputs and transforms
kubectl get compositions -o json | python3 -c "
import json,sys
comps = json.load(sys.stdin)['items']
for c in comps:
    meta = c['metadata']
    print(f\"Composition: {meta['name']}\")
    for resource in c.get('spec',{}).get('resources',[]):
        patches = resource.get('patches',[])
        for patch in patches:
            if patch.get('type') in ['FromCompositeFieldPath', 'CombineFromComposite']:
                from_field = patch.get('fromFieldPath','?')
                to_field = patch.get('toFieldPath','?')
                print(f\"  Patch: claim.{from_field} → resource.{to_field}\")
                # User-controlled fields flowing into resource specs
"

# Test: can a user set a claim field that overrides security controls?
# Example: Composition maps claim.spec.region to resource.spec.forProvider.region
# If region is not validated, attacker can provision in any region (including cheaper/less-monitored)

# Test: inject unexpected values into claim parameters
kubectl apply -f - </dev/null | head -10

# CompositeResourceDefinitions (XRDs) define what fields are available in claims
# If XRD allows setting providerConfigRef in the claim, users can choose which credentials to use

# Check XRDs for user-settable providerConfigRef
kubectl get compositeresourcedefinitions -o json | python3 -c "
import json,sys
xrds = json.load(sys.stdin)['items']
for xrd in xrds:
    meta = xrd['metadata']
    schema = xrd.get('spec',{}).get('versions',[{}])[0].get('schema',{})
    openapi = schema.get('openAPIV3Schema',{})
    spec_props = openapi.get('properties',{}).get('spec',{}).get('properties',{})
    if 'providerConfigRef' in spec_props:
        print(f\"[!] XRD allows providerConfigRef in claim: {meta['name']}\")
    if 'writeConnectionSecretToRef' in spec_props:
        print(f\"XRD allows custom connection secret namespace: {meta['name']}\")
        # User can redirect connection secret to their namespace
"

# Test: claim a composite resource with custom providerConfigRef
kubectl apply -f - <

# Crossplane writes connection details (credentials) to Kubernetes secrets
# These secrets contain database passwords, access keys, connection strings

# Find all connection secrets created by Crossplane
kubectl get managed --all-namespaces -o json 2>/dev/null | python3 -c "
import json,sys
try:
    resources = json.load(sys.stdin)['items']
    for r in resources:
        conn_ref = r.get('spec',{}).get('writeConnectionSecretToRef',{})
        if conn_ref:
            print(f\"{r['metadata']['name']} writes to: {conn_ref.get('namespace','?')}/{conn_ref.get('name','?')}\")
except: pass
"

# Check connection secret contents (reveals cloud resource credentials)
kubectl get secret CONN_SECRET_NAME -n NAMESPACE -o json | python3 -c "
import json,sys,base64
secret = json.load(sys.stdin)
for key, val in secret.get('data',{}).items():
    decoded = base64.b64decode(val).decode('utf-8','ignore')
    print(f'{key}: {decoded[:50]}...')
# Common keys: endpoint, port, username, password, attribute.account_id
"

# Check if connection secrets are accessible by unauthorized principals
kubectl auth can-i get secrets \
  --as=system:serviceaccount:developer:developer-sa \
  -n crossplane-system