Crypto & Auth

1
[auth]
2
provider = "r2r"
3
access_token_minutes_lifetime = 60
4
access_token_days_lifetime = 7
5
require_authentication = true
6
require_email_verification = false
7
default_admin_email = "[email protected]"
8
default_admin_password = "change_me_immediately"

1
[crypto]
2
provider = "bcrypt"
3
salt_rounds = 12

$
r2r generate-private-key

$
export R2R_SECRET_KEY=your_generated_key

1
from r2r import R2RClient, UserCreate
2
3
client = R2RClient("http://localhost:7272")
4
5
result = client.register([email protected], secure_password123)
6
print(f"Registration Result: {result}")

1
login_result = client.login("[email protected]", "secure_password123")
2
client.access_token = login_result['results']['access_token']['token']
3
client.refresh_token = login_result['results']['refresh_token']['token']

1
user = client.user()
2
print(f"Authenticated User Info: {user}")

1
refresh_result = client.refresh_access_token()
2
client.access_token = refresh_result['results']['access_token']['token']

1
logout_result = client.logout()
2
print(f"Logout Result: {logout_result}")

1
from r2r.base import AuthProvider, AuthConfig
2
from r2r.abstractions.user import User, UserCreate, Token, TokenData
3
from typing import Dict
4
5
class CustomAuthProvider(AuthProvider):
6
    def __init__(self, config: AuthConfig):
7
        super().__init__(config)
8
        # Initialize any custom attributes or connections here
9
10
    def create_access_token(self, data: dict) -> str:
11
        # Implement custom access token creation logic
12
        pass
13
14
    def create_refresh_token(self, data: dict) -> str:
15
        # Implement custom refresh token creation logic
16
        pass
17
18
    def decode_token(self, token: str) -> TokenData:
19
        # Implement custom token decoding logic
20
        pass
21
22
    def user(self, token: str) -> User:
23
        # Implement custom user info retrieval logic
24
        pass
25
26
    def get_current_active_user(self, current_user: User) -> User:
27
        # Implement custom active user validation logic
28
        pass
29
30
    def register(self, user: UserCreate) -> Dict[str, str]:
31
        # Implement custom user registration logic
32
        pass
33
34
    def verify_email(self, verification_code: str) -> Dict[str, str]:
35
        # Implement custom email verification logic
36
        pass
37
38
    def login(self, email: str, password: str) -> Dict[str, Token]:
39
        # Implement custom login logic
40
        pass
41
42
    def refresh_access_token(self, user_email: str, refresh_access_token: str) -> Dict[str, str]:
43
        # Implement custom token refresh logic
44
        pass
45
46
    async def auth_wrapper(self, auth: Optional[HTTPAuthorizationCredentials] = Security(security)) -> User:
47
        # You can override this method if you need custom authentication wrapper logic
48
        return await super().auth_wrapper(auth)
49
50
    # Add any additional custom methods as needed
51
    async def custom_auth_method(self, ...):
52
        # Implement custom authentication logic
53
        pass

1
from r2r.base import AuthProvider, AuthConfig
2
from r2r.abstractions.user import User, UserCreate, Token, TokenData
3
import some_oauth_library  # Replace with actual OAuth library
4
5
class OAuthAuthProvider(AuthProvider):
6
    def __init__(self, config: AuthConfig):
7
        super().__init__(config)
8
        self.oauth_client = some_oauth_library.Client(
9
            client_id=config.oauth_client_id,
10
            client_secret=config.oauth_client_secret
11
        )
12
13
    async def login(self, email: str, password: str) -> Dict[str, Token]:
14
        # Instead of password-based login, initiate OAuth flow
15
        auth_url = self.oauth_client.get_authorization_url()
16
        # Return auth_url or handle redirect as appropriate for your app
17
        pass
18
19
    async def oauth_callback(self, code: str) -> Dict[str, Token]:
20
        # Handle OAuth callback
21
        token = await self.oauth_client.get_access_token(code)
22
        user_data = await self.oauth_client.get_user_info(token)
23
24
        # Map external user data to R2R's user model
25
        r2r_user = self._map_oauth_user_to_r2r_user(user_data)
26
27
        # Create and return R2R tokens
28
        access_token = self.create_access_token({"sub": r2r_user.email})
29
        refresh_token = self.create_refresh_token({"sub": r2r_user.email})
30
        return {
31
            "access_token": Token(token=access_token, token_type="access"),
32
            "refresh_token": Token(token=refresh_token, token_type="refresh"),
33
        }
34
35
    def _map_oauth_user_to_r2r_user(self, oauth_user_data: dict) -> User:
36
        # Map OAuth user data to R2R User model
37
        return User(
38
            email=oauth_user_data["email"],
39
            # ... map other fields as needed
40
        )
41
42
    # Implement other required methods...

1
from r2r import R2R
2
from r2r.base import AuthConfig
3
from .custom_auth import OAuthAuthProvider
4
5
auth_config = AuthConfig(
6
    provider="custom_oauth",
7
    oauth_client_id="your_client_id",
8
    oauth_client_secret="your_client_secret",
9
    # ... other config options
10
)
11
12
r2r = R2R(
13
    auth_provider=OAuthAuthProvider(auth_config),
14
    # ... other R2R configuration
15
)

1
from fastapi import APIRouter, Depends
2
from r2r import R2R
3
4
router = APIRouter()
5
6
@router.get("/login")
7
async def login():
8
    return await r2r.auth_provider.login(None, None)  # Initiate OAuth flow
9
10
@router.get("/oauth_callback")
11
async def oauth_callback(code: str):
12
    return await r2r.auth_provider.oauth_callback(code)