Migration Guide: Legacy Auth to IAM¶
This guide explains how to migrate from the legacy authentication stack (g.current_user, g.actor, flask-login, custom auth decorators) to the new IAM stack (current_auth_context, AuthContextProvider, ABAC access policies).
Overview¶
| Legacy Stack | New Stack |
|---|---|
g.current_user / g.actor |
current_auth_context |
flask-login @login_required |
Auth context providers |
| Custom auth decorators | @enforce_policy with ABAC |
AuthorizationStrategy classes |
@enforce_policy with ABAC |
| Manual actor tracking | TransactionAuthContext audit trail |
When to migrate:
- New code should always use the new stack
- Migrate existing code when touching it for other reasons
- Full app migration should follow the steps below in order
Step 1: Setup TransactionAuthContext (Audit Trail)¶
Create an app-specific TransactionAuthContext model to enable audit trails linking database changes to authentication context.
1.1 Create the TransactionAuthContext model¶
from functools import partial
from sqlalchemy import Integer
from sqlalchemy.orm import Mapped, mapped_column
from shared.models.versioning.base import (
TransactionAuthContextBase,
default_real_principal_id,
)
class MyAppTransactionAuthContext(TransactionAuthContextBase):
"""Stores auth context for database transactions in my_app."""
schema = "my_app" # Your app's schema
# Add columns for each principal type your app uses
real_alaner_id: Mapped[int | None] = mapped_column(
Integer,
nullable=True,
index=True,
default=partial(default_real_principal_id, Alaner),
)
# Add relationship for convenient access
real_alaner: Mapped["Alaner | None"] = relationship(
"Alaner",
foreign_keys=[real_alaner_id],
lazy="joined",
)
@property
def actor(self) -> Alaner | None:
"""Returns the actor who made the change."""
return self.real_alaner
1.2 Create the database migration¶
Generate a migration to add the transaction_auth_context table:
The migration should create:
- The
transaction_auth_contexttable with your principal columns - An
auth_context_idforeign key column on the existingtransactiontable
1.3 Link TransactionAuthContext to Transaction¶
Update your app's Transaction model to reference TransactionAuthContext:
from sqlalchemy import ForeignKey
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
class MyAppTransaction(TransactionBase):
transaction_auth_context_cls = MyAppTransactionAuthContext
auth_context_id: Mapped[uuid.UUID | None] = mapped_column(
UUID(as_uuid=True),
ForeignKey(MyAppTransactionAuthContext.id),
index=True,
)
auth_context: Mapped[MyAppTransactionAuthContext | None] = relationship(
MyAppTransactionAuthContext,
foreign_keys=[auth_context_id],
)
Step 2: Make Principal Classes Implement AuthPrincipal¶
Your principal classes (users, employees, service accounts) need to implement AuthPrincipal for the new auth system to recognize them.
from shared.iam.auth_principal import AuthPrincipal, AuthPrincipalType
class Alaner(BaseModel, AuthPrincipal):
"""Alan employee model."""
# Configure auth principal behavior
__auth_principal_type__ = AuthPrincipalType.back_office
__identifier_column__ = "email" # Column used for lookups by auth providers
# ... existing model code ...
AuthPrincipalType Values¶
| Type | Description |
|---|---|
back_office |
Internal employees (Alaners) |
front_office |
End users (members, customers) |
service_account |
Automated service accounts |
Step 3: Configure Auth Context Providers¶
Auth context providers extract authentication from requests and set up current_auth_context.
3.1 Choose your providers¶
| Provider | Use Case |
|---|---|
ZeroTrustAuthContextProvider |
Cloudflare Zero Trust (backoffice) |
ServiceAccountAuthContextProvider |
Google service accounts |
DevAuthContextProvider |
Local development |
AnonymousAuthContextProvider |
Public endpoints (fallback) |
WebhookAuthContextProvider |
Webhook authentication |
3.2 Configure at app level¶
from shared.iam.auth_context.providers.dev import DevAuthContextProvider
from shared.iam.auth_context.providers.zero_trust import ZeroTrustAuthContextProvider
def create_app():
app = Flask(__name__)
# Choose provider based on environment
if is_development_mode():
auth_provider = DevAuthContextProvider(Alaner)
else:
auth_provider = ZeroTrustAuthContextProvider(Alaner)
# Register with your API
api = CustomApi(app)
api.register_default_auth_context_providers([auth_provider])
return app
3.3 Blueprint-level customization¶
Individual blueprints can override the default providers:
from shared.api.custom_smorest_blueprint import CustomBlueprint
from shared.iam.auth_context.providers.anonymous import AnonymousAuthContextProvider
# Public blueprint allowing unauthenticated access
public_blueprint = CustomBlueprint(
"public",
__name__,
auth_context_providers=[
ZeroTrustAuthContextProvider(User),
AnonymousAuthContextProvider(), # Fallback for public access
],
)
Step 4: Bridge Legacy Auth to current_auth_context¶
Before migrating endpoints, ensure current_auth_context is set alongside g.current_user in legacy authentication code. This allows both systems to work consistently during migration.
4.1 Update legacy authentication to set both¶
Find where legacy auth globals are set and add set_auth_context:
# Before
from shared.iam.helpers import set_auth_globals_for_regular_user
def authenticate_user():
user = validate_token_and_get_user()
set_auth_globals_for_regular_user(user)
# After - set both for consistency
from shared.iam.helpers import set_auth_context, set_auth_globals_for_regular_user
def authenticate_user():
user = validate_token_and_get_user()
# Legacy - keep for backwards compatibility
set_auth_globals_for_regular_user(user)
# New - enables current_auth_context
set_auth_context(real_principal=user)
4.2 Handle impersonation¶
If your app supports impersonation, set both:
# Before
from shared.iam.helpers import set_auth_globals_for_impersonated_user
def authenticate_with_impersonation():
alaner = get_authenticated_alaner()
impersonated_user = get_impersonated_user()
set_auth_globals_for_impersonated_user(alaner, impersonated_user, ImpersonationMode.read_write)
# After - set both for consistency
from shared.iam.helpers import (
set_auth_context,
set_auth_globals_for_impersonated_user,
ImpersonationMode,
)
def authenticate_with_impersonation():
alaner = get_authenticated_alaner()
impersonated_user = get_impersonated_user()
# Legacy
set_auth_globals_for_impersonated_user(alaner, impersonated_user, ImpersonationMode.read_write)
# New
set_auth_context(
real_principal=alaner,
effective_principal=impersonated_user,
impersonation_mode=ImpersonationMode.read_write,
)
This ensures code using current_auth_context works correctly even on legacy endpoints.
Step 5: Migrate from Legacy Auth¶
5.1 Replace g.current_user / g.actor¶
# After
from shared.iam.helpers import current_auth_context
def my_endpoint():
# Type-safe access
user = current_auth_context.real_principal_as(Alaner)
user_id = user.id
5.2 Replace @login_required + custom decorators with ABAC¶
Flask-login apps only
@login_required is specific to apps using Flask-login. Skip this if your app uses a different authentication mechanism.
# Before
from flask_login import login_required
from my_app.auth import my_custom_auth_permitted_for
@login_required
@my_custom_auth_permitted_for({EmployeePermission.manage_users})
def admin_endpoint():
...
# After
from shared.iam.abac.access_policy import enforce_policy
from shared.iam.abac.backoffice import BackofficePermissionAccessPolicy
from shared.models.enums.employee_permission import EmployeePermission
@enforce_policy(BackofficePermissionAccessPolicy.permitted_for(
EmployeePermission.manage_users
))
def admin_endpoint():
...
5.3 Replace authentication checks¶
# After
if current_auth_context.is_anonymous:
abort(403)
# Or use a policy
@enforce_policy(AuthenticatedPolicy)
def protected_endpoint():
...
5.4 Handle impersonation¶
During impersonation, real_principal and effective_principal differ:
from shared.iam.helpers import current_auth_context
# The Alaner performing the action
actor = current_auth_context.real_principal_as(Alaner)
# The user being impersonated (or same as actor if not impersonating)
subject = current_auth_context.effective_principal_as(User)
# Check if currently impersonating
if current_auth_context.is_impersonated:
print(f"{actor.email} is acting as {subject.email}")
See Authentication Context for more details on impersonation.
5.5 Replace AuthorizationStrategies with ABAC policies¶
The legacy BaseAuthorizationStrategy classes are replaced by ABAC access policies.
Strategy mapping¶
| Legacy Strategy | New Approach |
|---|---|
OpenStrategy |
AnonymousAuthContextProvider (no policy needed) |
AuthenticatedStrategy |
Auth context providers (authentication is implicit) |
AuthenticatedWithCustomAuthorizationStrategy |
Custom AccessPolicy with @enforce_policy |
AlanerAdminStrategy(permitted_for={...}) |
BackofficePermissionAccessPolicy.permitted_for(...) |
OwnerOnlyStrategy |
Custom AccessPolicy checking ownership |
Migrating AlanerAdminStrategy¶
# Before (in controller)
from shared.iam.authorization import AlanerAdminStrategy
class MyController(BaseController):
@action_route(
"/admin-action",
auth_strategy=AlanerAdminStrategy(permitted_for={EmployeePermission.manage_users}),
)
def admin_action(self):
...
# After (with CustomBlueprint)
from shared.iam.abac.access_policy import enforce_policy
from shared.iam.abac.backoffice import BackofficePermissionAccessPolicy
@blueprint.route("/admin-action")
@enforce_policy(BackofficePermissionAccessPolicy.permitted_for(
EmployeePermission.manage_users
))
def admin_action():
...
Migrating AuthenticatedWithCustomAuthorizationStrategy¶
# Before
from shared.iam.authorization import (
AuthenticatedWithCustomAuthorizationStrategy,
custom_authorization,
)
class MyController(BaseController):
@action_route(
"/resource/<int:resource_id>",
auth_strategy=AuthenticatedWithCustomAuthorizationStrategy(),
)
def get_resource(self, resource_id):
self._check_resource_access(resource_id)
...
@custom_authorization
def _check_resource_access(self, resource_id):
resource = Resource.query.get(resource_id)
return resource and resource.owner_id == g.current_user.id
# After
from shared.iam.abac.access_policy import AccessPolicy, enforce_policy
from shared.iam.helpers import current_auth_context
class ResourceOwnerPolicy(AccessPolicy):
"""User must own the resource"""
policy_id = "resource-owner"
@classmethod
def evaluate(cls, resource_id: int) -> bool:
if current_auth_context.is_anonymous:
return False
user = current_auth_context.real_principal_as(User)
resource = Resource.query.get(resource_id)
return resource is not None and resource.owner_id == user.id
@blueprint.route("/resource/<int:resource_id>")
@enforce_policy(ResourceOwnerPolicy)
def get_resource(resource_id: int):
...
Migrating OwnerOnlyStrategy¶
Use ControllerOwnershipBasedAccessPolicy to reuse existing can_read/can_write logic from controllers:
# Before
from shared.iam.authorization import OwnerOnlyStrategy
class UserController(BaseController):
@action_route(
"/user/<int:id>",
auth_strategy=OwnerOnlyStrategy(
owner_bypass_permitted_for={EmployeePermission.admin}
),
)
def update_user(self, id):
...
def can_write(self, user, resource_id):
return user.id == resource_id
# After - reuse existing can_read/can_write via factory
from shared.iam.abac.access_policy import enforce_policy, or_
from shared.iam.abac.backoffice import BackofficePermissionAccessPolicy
from shared.iam.abac.controller_ownership import ControllerOwnershipBasedAccessPolicy
@blueprint.route("/user/<int:id>", methods=["PUT"])
@enforce_policy(or_(
ControllerOwnershipBasedAccessPolicy.from_controller(UserController),
BackofficePermissionAccessPolicy.permitted_for(EmployeePermission.admin),
))
def update_user(id: int):
...
The factory automatically uses can_read for GET requests and can_write for mutating requests, matching the legacy OwnerOnlyStrategy behavior.
5.6 Delete legacy auth code¶
After migration, remove:
@login_requireddecorators- Custom auth decorator files
- flask-login setup from app initialization
g.current_user/g.actorusagesAuthorizationStrategyclasses and@custom_authorizationdecorators
Step 6: Use CustomBlueprint¶
Replace create_blueprint() with CustomBlueprint for full IAM integration:
# After
from shared.api.custom_smorest_blueprint import CustomBlueprint
blueprint = CustomBlueprint(
"my_blueprint",
__name__,
# Optional: blueprint-specific auth providers
auth_context_providers=[...],
# Optional: default policies for all routes
default_access_policies=[...],
)
Migration Checklist¶
- [ ] Create
TransactionAuthContextmodel for your app - [ ] Create database migration for
transaction_auth_contexttable - [ ] Link
TransactionAuthContextto yourTransactionmodel - [ ] Principal classes implement
AuthPrincipalwith__identifier_column__ - [ ] Auth context providers configured at app level
- [ ] Blueprints use
CustomBlueprint - [ ] Replace
g.current_user/g.actorwithcurrent_auth_context - [ ] Replace
@login_required+ custom decorators with@enforce_policy - [ ] Replace
AuthorizationStrategyclasses with ABAC policies - [ ] Delete legacy auth code (custom decorators, flask-login setup, AuthorizationStrategy classes)
- [ ] Verify audit trail works (check
transaction_auth_contextrecords)
Troubleshooting¶
"No auth provider matched the request"¶
Ensure exactly one provider returns True from will_handle_request(). Add AnonymousAuthContextProvider as a fallback if public access is needed.
"current_auth_context is anonymous in tests"¶
Use the set_auth_context helper in test fixtures:
from shared.iam.helpers import set_auth_context
@pytest.fixture
def authenticated_user(db):
user = UserFactory.create()
set_auth_context(real_principal=user)
yield user
TransactionAuthContext not being created¶
Verify:
- Your
Transactionclass hastransaction_auth_context_clsset - The migration added the
auth_context_idcolumn totransaction - Auth context is set before database operations