initial package

.gitignore

@@ -0,0 +1,8 @@
+__pycache__/
+*.py[cod]
+dist/
+services/__pycache__/
+services/*.py[cod]
+tests/__pycache__/
+tests/*.py[cod]
+routes/__pycache__/

README.md

@@ -1 +1,113 @@
-# auth-core-flask
+# Flycatch Auth
+
+Flycatch Auth is a Python authentication package that provides JWT-based authentication with grant-based access control for both Flask and FastAPI applications.
+
+## Features
+
+- **JWT authentication**:Supports access and refresh tokens with customizable expiry times.
+
+- **Grant-based access control**
+
+- **Works with Flask and FastAPI**
+
+- **Simple integration with existing user services**
+
+## Installation
+
+Install flycatch_auth using pip:
+
+```bash
+pip install flycatch-auth
+```
+
+## Usage
+
+```python
+
+from flask import Flask, request, jsonify
+from flycatch_auth import auth, AuthCoreJwtConfig
+
+app = Flask(__name__)
+
+class Userservice(IdentityService):
+    def load_user(self,username: str) -> Identity:
+        user = user_db.get_users()
+        return {
+                "id": "",
+                "username": "",
+                "password": "",
+                "grants": [list(map(lambda user: user.permission.name, users))] # read_user
+                }
+
+jwt_config = AuthCoreJwtConfig(
+    enable=True,
+    secret="mysecret",
+    expiresIn="2h",
+    refresh=True,
+    prefix="/auth/jwt",
+)
+
+auth.init_app(
+    app=app,
+    user_service=MockUserService(),
+    credential_checker=lambda input, user: input == user,
+    jwt=jwt_config,
+)
+
+
+@app.route("/me")
+@auth.verify()
+def get_curr_user():
+    return jsonify({"message": "Access granted"}), 200
+
+if __name__ == "__main__":
+    app.run(debug=True)
+```
+
+This snippet demonstrates how to configure and initialize Auth using MockUserService and JWT-based authentication
+
+## Configuration Options
+
+### JWT Configuration
+
+```python
+jwt_config = {
+    "enabled": True,
+    "secret": "your-jwt-secret",
+    "expiresIn": "8h",  # Expiry time for JWT
+    "refreshToken": True,  # Enable refresh tokens
+    "prefix": "/auth/jwt",  # Prefix for JWT-related routes
+}
+```
+### User Service
+
+```python
+class UserService(IdentityService):
+    def load_user(self, username: str)-> Identity:
+        # users = users.get_db()
+        # your logic for implemention
+        return {
+            "id": "1",
+            "username": "testuser",
+            "password": "password123",
+            "grants": ["read_user"],
+        }
+```
+
+### Password Checker
+
+```python
+credential_checker=lambda input, user: input == user["password"],
+```
+
+## License
+
+This project is licensed under the GPL-3.0 License.
+
+---
+
+For more details and advanced use cases, visit the [GitHub repository](#) or contact the project maintainers.
+
+```
+
+```

flycatch_auth/__init__.py

@@ -0,0 +1,8 @@
+from .auth import AuthCore 
+from .model_types import Identity, IdentityService, CredentialChecker
+from .services import AuthCoreJwtConfig
+
+auth = AuthCore()
+
+# It enables top-level imports like `from flycatch_auth import auth, AuthCoreJwtConfig, JwtService`
+__all__ = ["auth", "Identity", "IdentityService", "AuthCoreJwtConfig","CredentialChecker"]

flycatch_auth/auth.py

@@ -0,0 +1,53 @@
+import logging
+# from flask_session import Session
+# from authlib.integrations.flask_client import OAuth
+from .model_types import IdentityService, CredentialChecker
+from .services import AuthCoreJwtConfig, JwtAuthService
+from .routes import create_jwt_routes
+from .middleware import verify_request
+logger = logging.getLogger("auth_core")
+
+
+class AuthCore:
+    def __init__(self):
+        self.app = None
+        self.jwt = None
+        self.session = None
+        self.user_service = None
+        self.credential_checker = None
+        self.auth_service = None
+
+    def init_app(
+        self,
+        app,
+        user_service: IdentityService,
+        credential_checker: CredentialChecker,
+        jwt: AuthCoreJwtConfig,
+    ):
+        """Initialize authentication with JWT if enabled."""
+        self.app = app
+        self.jwt = jwt
+        self.user_service = user_service
+        self.credential_checker = credential_checker
+
+        if jwt and jwt.get("enable"):
+            # Initialize JWT authentication service
+            self.auth_service = JwtAuthService(
+                user_service, credential_checker, jwt)
+
+            # Set up JWT authentication routes
+            create_jwt_routes(app, jwt, user_service, credential_checker)
+
+        # if session and session.enable:
+        #     # Initialize session-based authentication service
+        #     self.auth_service = SessionAuthService(
+        #         user_service, credential_checker, session)
+
+        return self.auth_service
+
+    def verify(self):
+        """Middleware to verify authentication."""
+        return verify_request(self.auth_service)
+
+
+auth = AuthCore()

flycatch_auth/middleware.py

@@ -0,0 +1,41 @@
+import jwt
+import logging
+from functools import wraps
+from flask import request, jsonify
+from .model_types import api_response
+logger = logging.getLogger("auth_core")
+
+
+def verify_request(auth_service):
+    """Middleware function to verify authentication."""
+
+    def decorator(f):
+        @wraps(f)
+        def wrapper(*args, **kwargs):
+            auth_header = request.headers.get("Authorization")
+
+            if not auth_header:
+                logger.warning(
+                    "Unauthorized access attempt: Missing Authorization header")
+                return api_response(401, "Unauthorized access attempt",False)
+            if auth_service.jwt_config.get("enable"):
+                try:
+                    token = auth_header.split(" ")[1]  # Extract the token
+                    decoded_token = jwt.decode(
+                        token, auth_service.jwt_config.get('secret'), algorithms=["HS256"])
+
+                    logger.info(
+                        f"User {decoded_token['username']} authenticated successfully")
+                    request.user = decoded_token  # Attach user info to request
+                    return f(*args, **kwargs)
+
+                except jwt.ExpiredSignatureError:
+                    logger.warning("Token expired")
+                    return api_response(403, "Token expired", False)
+                except jwt.InvalidTokenError:
+                    logger.warning("Invalid token")
+                    return api_response(401, "Invalid token", False)
+
+        return wrapper
+
+    return decorator

flycatch_auth/model_types.py

@@ -0,0 +1,68 @@
+from abc import ABC, abstractmethod
+from typing import List, Optional, TypedDict
+from datetime import datetime
+
+
+class Identity:
+    """Manage user authentication and authorization."""
+
+    def __init__(self, user_id: str, username: str, grants: List[str]):
+        self.id = user_id
+        self.username = username
+        self.grants = grants
+
+    def has_grant(self, grant: str) -> bool:
+        """Check if the identity has a specific grant."""
+        return grant in self.grants
+
+
+class IdentityService(ABC):
+    @abstractmethod
+    def load_user(self, username: str) -> Optional[Identity]:
+        """Override this method to fetch a user from DB."""
+        pass
+
+
+class UserType(TypedDict):
+    id: str
+    username: str
+    password: str
+
+
+def api_response(code: int, message: str, status: str, data=None):
+    """Format API response."""
+    return {
+        "code": code,
+        "timestamp": datetime.utcnow().isoformat() + "Z",
+        "message": message,
+        "status": status,
+        "data": data or {},
+    }
+
+
+class CredentialChecker:
+    def __init__(self):
+        pass
+
+    def verify(self, input_password: str, user_password: str) -> bool:
+        """Checks if the provided password matches the stored password."""
+        return input_password == user_password
+
+
+class SessionConfig(TypedDict, total=False):
+    session_key: str
+    session_expiration: int
+    enabled: bool
+    prefix: str
+    secret: str
+    resave: bool
+    saveUninitialized: bool
+    cookie: dict
+
+
+class AuthCoreJwtConfig(TypedDict, total=False):
+    enable: bool
+    secret: str
+    expiresIn: str
+    refresh: bool
+    prefix: str

flycatch_auth/routes/__init__.py

@@ -0,0 +1,3 @@
+from .jwt_routes import create_jwt_routes
+
+__all__ = ["create_jwt_routes"]

flycatch_auth/routes/jwt_routes.py

@@ -0,0 +1,36 @@
+from flask import Blueprint, request, jsonify
+import logging
+from ..services import JwtAuthService
+
+
+logger = logging.getLogger(__name__)
+
+
+def create_jwt_routes(app, config, user_service, credential_checker):
+    router = Blueprint("jwt_auth", __name__)
+    prefix = config.get("prefix") or "/auth/jwt"
+
+    jwt_auth = JwtAuthService(user_service, credential_checker, config)
+
+    @router.route(f"{prefix}/login", methods=["POST"])
+    def login():
+        """Login route to generate access and refresh tokens"""
+        data = request.json
+        username, password = data.get("username"), data.get("password")
+        return jsonify(jwt_auth.login(username=username, password=password))
+
+    @router.route(f"{prefix}/refresh", methods=["POST"])
+    def refresh():
+        """Refresh access token using a valid refresh token"""
+        auth_header = request.headers.get("Authorization")
+        logger.info("Refresh token attempt received")
+
+        if not auth_header:
+            logger.warning("Refresh token missing in request")
+            return jsonify({"error": "Refresh token is required"}), 400
+
+        refresh_token = auth_header.split(" ")[1]
+        response = jwt_auth.refresh(refresh_token)
+        return jsonify(response), 200 if "access_token" in response else 403
+
+    app.register_blueprint(router)

flycatch_auth/services/__init__.py

@@ -0,0 +1,5 @@
+from .base import AuthService
+from .jwt_services import JwtAuthService, AuthCoreJwtConfig
+# from .session_service import SessionAuthService  # Uncomment when session auth is implemented
+
+__all__ = ["AuthService", "JwtAuthService","AuthCoreJwtConfig"]

flycatch_auth/services/base.py

@@ -0,0 +1,25 @@
+from abc import ABC, abstractmethod
+from typing import Optional, Dict
+
+
+class AuthService(ABC):
+    """Abstract base class for authentication services."""
+
+    @abstractmethod
+    def authenticate(self, username: str, password: str) -> Optional[Dict]:
+        """Authenticate a user and return user data if successful."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def login(self, username: str, password: str) -> Dict:
+        """Handle user login and return authentication tokens or session details."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def refresh(self, refresh_token: str) -> Dict:
+        """Refresh authentication tokens (if applicable)."""
+        raise NotImplementedError
+
+
+
+