Implementing a RAC Provider in Authentik: Complete Technical Guide
Authentik's Remote Access Control (RAC) provider creates time-bound SSH connection tokens validated by outposts, enabling secure temporary server access without permanent credentials.
The RAC (Remote Access Control) provider in Authentik represents a sophisticated architecture for just-in-time SSH access. Unlike traditional VPN or bastion host approaches, RAC generates short-lived connection tokens through configurable authorization flows, bridging web-based identity management with low-level protocol access. This guide examines the complete implementation across Python Django models, ASGI consumers, REST API, and Go-based outpost components.
RAC Provider Architecture Overview
The RAC implementation spans three runtime environments: the Authentik core server (Python/Django), the outpost proxy (Go), and the target endpoints. According to the source code in goauthentik/authentik, this follows Authentik's standard provider pattern with additional complexity for real-time bidirectional communication.
Core Design Principles
- Token-based mediation: No long-lived credentials stored on client systems
- Flow-driven authorization: Multi-factor authentication and policy enforcement before access grant
- Outpost-mediated connections: All traffic proxied through validated, ephemeral tunnels
- Protocol abstraction: SSH support with extensibility for RDP and other protocols
Data Model Implementation in models.py
The foundation resides in authentik/providers/rac/models.py, defining three primary entities with hierarchical relationships.
RACProvider Model
The RACProvider class extends OutpostModel and Provider, inheriting deployment and application-binding capabilities:
# From authentik/providers/rac/models.py
class RACProvider(OutpostModel, Provider):
authorization_flow = models.ForeignKey(
Flow,
on_delete=models.CASCADE,
help_text="Flow used for authorizing this provider",
)
property_mappings = models.ManyToManyField(
RACPropertyMapping,
blank=True,
help_text="Property mappings for this provider",
)
default_endpoint = models.ForeignKey(
"Endpoint",
on_delete=models.SET_NULL,
null=True,
blank=True,
)
Key fields include:
authorization_flow: Mandatory reference to a Flow instance executing authentication and policy stagesproperty_mappings: Optional RAC-specific attribute mappings for protocol propertiesdefault_endpoint: Fallback destination when no explicit endpoint specified
Endpoint Model
The Endpoint class represents target resources with protocol-specific configuration:
# Conceptual structure from source analysis
class Endpoint(models.Model):
name = models.CharField(max_length=255)
provider = models.ForeignKey(RACProvider, on_delete=models.CASCADE)
protocol = models.CharField(
choices=Protocols.choices,
max_length=100,
)
address = models.CharField(max_length=255)
port = models.PositiveIntegerField(default=22)
host = models.ForeignKey(
"authentik_outposts.ServiceConnection",
on_delete=models.CASCADE,
)
Supported protocols (defined in Protocols enum) currently include SSH with planned extensibility.
ConnectionToken Model
The ConnectionToken implements temporal access constraints:
# Core fields from models.py
class ConnectionToken(models.Model):
token = models.UUIDField(primary_key=True, default=uuid4)
provider = models.ForeignKey(RACProvider, on_delete=models.CASCADE)
endpoint = models.ForeignKey(Endpoint, on_delete=models.CASCADE)
user = models.ForeignKey(get_user_model(), on_delete=models.CASCADE)
expiration = models.DateTimeField()
session_id = models.CharField(max_length=255)
Critical characteristics:
- UUID primary key with automatic generation
- Explicit expiration timestamp validated on every use
- Session linkage for traceability and revocation
API Layer: Serializers and ViewSets
The REST API implementation in authentik/providers/rac/api/ enables programmatic RAC management.
RACProviderSerializer
Located in authentik/providers/rac/api/providers.py:
from authentik.providers.rac.models import RACProvider
from authentik.core.api.providers import ProviderSerializer
class RACProviderSerializer(ProviderSerializer):
class Meta:
model = RACProvider
fields = ProviderSerializer.Meta.fields + [
"authorization_flow",
"property_mappings",
"default_endpoint",
"endpoints",
]
The serializer inherits from ProviderSerializer for consistent provider API behavior while exposing RAC-specific relationships.
RACPropertyMappingSerializer
From authentik/providers/rac/api/property_mappings.py:
class RACPropertyMappingSerializer(PropertyMappingSerializer):
class Meta:
model = RACPropertyMapping
fields = PropertyMappingSerializer.Meta.fields + [
"expression",
]
Property mappings enable dynamic attribute transformation using authentik's expression engine.
Provider Creation via API
import requests
# Create RAC provider programmatically
response = requests.post(
"https://authentik.example.com/api/v3/providers/rac/",
headers={"Authorization": "Bearer token"},
json={
"name": "Production SSH RAC",
"authorization_flow": "flow-uuid-here",
"property_mappings": ["mapping-uuid-1", "mapping-uuid-2"],
},
)
provider = response.json()
View Layer: Flow Integration and Interface
The authentik/providers/rac/views.py implements the critical bridge between web flows and protocol access.
RACStartView
This view initiates the RAC access sequence:
# From authentik/providers/rac/views.py
class RACStartView(AccessMixin, View):
def dispatch(self, request, *args, **kwargs):
# 1. Retrieve provider by application slug
provider = get_object_or_404(
RACProvider,
application__slug=kwargs["application_slug"]
)
# 2. Execute authorization flow if required
if not self.test_policy(request, provider):
return self.handle_no_permission()
# 3. Create ConnectionToken with configured TTL
token = ConnectionToken.objects.create(
provider=provider,
user=request.user,
expiration=now() + timedelta(
seconds=provider.token_validity_seconds
),
)
# 4. Redirect to RAC interface with token
return redirect(
reverse("authentik_providers_rac:interface")
+ f"?token={token.token}"
)
The view enforces policy evaluation before token generation, enabling contextual access decisions based on user attributes, time, or external signals.
RACFinalStage
Flow stage implementation completing RAC authorization:
class RACFinalStage(RedirectStage):
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
# Store token in flow context for downstream consumption
context["connection_token"] = self.executor.plan.context.get(
"connection_token"
)
return context
URL Routing and ASGI Registration
The authentik/providers/rac/urls.py multiplexes HTTP and WebSocket endpoints:
from django.urls import path
from authentik.providers.rac.views import (
RACStartView,
RACInterface,
)
from authentik.providers.rac.consumer_client import RACClientConsumer
from authentik.providers.rac.consumer_outpost import RACOutpostConsumer
urlpatterns = [
path(
"application/<slug:application_slug>/start/",
RACStartView.as_view(),
name="start",
),
path("interface/", RACInterface.as_view(), name="interface"),
]
websocket_urlpatterns = [
path("ws/rac/client/", RACClientConsumer.as_asgi()),
path("ws/rac/outpost/", RACOutpostConsumer.as_asgi()),
]
Outpost Communication: ASGI Consumers
Bidirectional WebSocket communication enables real-time tunnel establishment between browser, authentik server, and outpost.
RACClientConsumer
From authentik/providers/rac/consumer_client.py:
class RACClientConsumer(AsyncWebsocketConsumer):
async def connect(self):
# Validate ConnectionToken from query string
self.token = await self.get_valid_token(
self.scope["query_string"]
)
if not self.token:
await self.close(code=4001)
return
# Join token-specific channel group
await self.channel_layer.group_add(
f"rac_{self.token.token}",
self.channel_name,
)
await self.accept()
async def receive(self, text_data=None, bytes_data=None):
# Forward client data to outpost via channel layer
await self.channel_layer.group_send(
f"rac_outpost_{self.token.token}",
{
"type": "tunnel.data",
"data": bytes_data or text_data,
},
)
RACOutpostConsumer
From authentik/providers/rac/consumer_outpost.py:
class RACOutpostConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.outpost = await self.authenticate_outpost(
self.scope["headers"]
)
if not self.outpost:
await self.close(code=4002)
return
await self.accept()
async def tunnel_data(self, event):
# Receive from channel layer, send to outpost
await self.send(bytes_data=event["data"])
The channel layer mediates between client and outpost WebSockets, decoupling their lifecycles while maintaining ordered message delivery.
Go Outpost Implementation
The production tunneling logic resides in internal/outpost/rac/rac.go:
// Simplified structure from source analysis
type RacOutpost struct {
client *http.Client
akApiUrl string
token string
tunnels map[string]*Tunnel
}
func (r *RacOutpost) Start() error {
// Establish WebSocket connection to authentik core
ws, _, err := websocket.DefaultDialer.Dial(
r.akApiUrl+"/ws/rac/outpost/",
http.Header{"Authorization": []string{"Bearer " + r.token}},
)
if err != nil {
return fmt.Errorf("failed to connect to authentik: %w", err)
}
// Start tunnel manager
go r.manageTunnels(ws)
return nil
}
func (r *RacOutpost) handleNewConnection(connToken string) (*Tunnel, error) {
// Validate token with authentik API
token, err := r.validateToken(connToken)
if err != nil {
return nil, fmt.Errorf("token validation failed: %w", err)
}
// Establish connection to target endpoint
target, err := net.Dial(
token.Protocol,
fmt.Sprintf("%s:%d", token.Address, token.Port),
)
if err != nil {
return nil, fmt.Errorf("target connection failed: %w", err)
}
// Create bidirectional tunnel
tunnel := &Tunnel{
Token: token,
Client: ws,
Target: target,
}
go tunnel.Proxy()
return tunnel, nil
}
The Go outpost handles protocol-specific negotiations (SSH handshake, host key verification) while treating the authentik-mediated WebSocket as a reliable transport.
Configuration and Deployment
Creating Provider via Django Shell
from authentik.providers.rac.models import RACProvider, Endpoint, Protocols
from authentik.core.models import Application
from authentik.flows.models import Flow
# Prerequisites
auth_flow = Flow.objects.get(name="SSH Authorization")
outpost_connection = ServiceConnection.objects.get(name="Docker Local")
# Create provider
provider = RACProvider.objects.create(
name="Production RAC",
authorization_flow=auth_flow,
token_validity_seconds=3600, # 1 hour
)
# Define SSH endpoint
endpoint = Endpoint.objects.create(
provider=provider,
name="production-db-01",
protocol=Protocols.SSH,
address="10.0.1.15",
port=22,
host=outpost_connection,
)
# Bind to application
app = Application.objects.create(
name="Production Database Access",
slug="prod-db-ssh",
provider=provider,
)
Outpost Deployment Configuration
# docker-compose.yml fragment for RAC outpost
services:
rac-outpost:
image: ghcr.io/goauthentik/rac-outpost:latest
environment:
AUTHENTIK_HOST: https://authentik.example.com
AUTHENTIK_TOKEN: ${OUTPOST_TOKEN}
AUTHENTIK_INSECURE: "false"
ports:
- "3389:3389" # RDP (future)
- "2222:2222" # SSH proxy
Testing Implementation
The test suite in authentik/providers/rac/tests/ validates all components:
Model Tests (test_models.py)
class TestRACProvider(TestCase):
def test_token_expiration(self):
provider = RACProviderFactory()
token = ConnectionToken.objects.create(
provider=provider,
expiration=now() - timedelta(seconds=1),
)
self.assertTrue(token.is_expired())
def test_endpoint_protocol_validation(self):
with self.assertRaises(ValidationError):
Endpoint.objects.create(
protocol="INVALID_PROTOCOL", # Should fail
)
API Tests (test_api.py, test_connection_tokens_api.py, test_endpoints_api.py)
class TestRACProviderAPI(APITestCase):
def test_provider_list(self):
response = self.client.get("/api/v3/providers/rac/")
self.assertEqual(response.status_code, 200)
def test_start_flow_action(self):
provider = RACProviderFactory()
response = self.client.post(
f"/api/v3/providers/rac/{provider.pk}/start/"
)
self.assertEqual(response.status_code, 302) # Redirect to flow
View Tests (test_views.py)
class TestRACStartView(TestCase):
def test_policy_denial(self):
# Create provider with deny-all policy
response = self.client.get(
reverse("authentik_providers_rac:start", kwargs={
"application_slug": "blocked-app"
})
)
self.assertEqual(response.status_code, 403)
Integration with Frontend Components
Authentik's TypeScript API client exposes RAC operations:
import { RACProvidersApi, RACApi } from "@goauthentik/api";
const providersApi = new RACProvidersApi(new Configuration({
basePath: "https://authentik.example.com",
accessToken: "token",
}));
// List RAC providers
const providers = await providersApi.racProvidersList();
// Start RAC session (returns redirect URL)
const startResponse = await providersApi.racProvidersStartCreate({
uuid: provider.uuid,
});
// Navigate to RAC interface
window.location.href = startResponse.redirect;
Summary
Implementing a RAC provider in Authentik requires coordinated configuration across multiple subsystems:
- Define the data model using
RACProvider,Endpoint, andConnectionTokeninmodels.py - Expose management APIs through serializers in
api/providers.pyandapi/property_mappings.py - Implement flow integration via
RACStartViewandRACFinalStageinviews.py - Configure routing in
urls.pyfor both HTTP and WebSocket endpoints - Deploy outpost infrastructure running the Go implementation from
internal/outpost/rac/rac.go - Establish real-time channels through
RACClientConsumerandRACOutpostConsumer
The architecture prioritizes security through temporality: no persistent credentials, explicit expiration, comprehensive audit logging, and policy-enforced access decisions.
Frequently Asked Questions
What protocols does the RAC provider support?
The RAC provider currently implements SSH as the primary protocol, with the Protocols enum in models.py designed for extensibility. The Go outpost in internal/outpost/rac/rac.go handles SSH-specific handshake and host key verification. RDP support is planned for future releases based on the protocol abstraction in the endpoint model.
How long do RAC connection tokens remain valid?
Token validity is configurable per provider through the token_validity_seconds field (default typically 3600 seconds). The ConnectionToken model enforces expiration via the expiration datetime field, validated on every use in both the WebSocket consumer and Go outpost. Expired tokens return HTTP 401 or WebSocket close code 4001.
Can RAC integrate with existing SSH key infrastructure?
Yes. According to website/docs/add-secure-apps/providers/rac/rac-public-key.md, RAC supports public key authentication delegation. The property mapping system (RACPropertyMapping) can inject user-specific public keys into SSH certificate generation or forward them through the outpost tunnel for host-side validation.
What happens when multiple outposts serve the same RAC provider?
The OutpostModel inheritance enables automatic load balancing and failover. When multiple outposts register for a provider, authentik distributes connection requests based on outpost health and capacity. The ServiceConnection field on Endpoint directs specific endpoints to designated outpost clusters for geographic or network segmentation.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →