Tirando env do commit

2025-06-21 16:19:21 +00:00
parent e51648489b
commit 5feb9db6c0
2549 changed files with 397274 additions and 0 deletions
@@ -0,0 +1,15 @@
+from .api_key import APIKeyCookie as APIKeyCookie
+from .api_key import APIKeyHeader as APIKeyHeader
+from .api_key import APIKeyQuery as APIKeyQuery
+from .http import HTTPAuthorizationCredentials as HTTPAuthorizationCredentials
+from .http import HTTPBasic as HTTPBasic
+from .http import HTTPBasicCredentials as HTTPBasicCredentials
+from .http import HTTPBearer as HTTPBearer
+from .http import HTTPDigest as HTTPDigest
+from .oauth2 import OAuth2 as OAuth2
+from .oauth2 import OAuth2AuthorizationCodeBearer as OAuth2AuthorizationCodeBearer
+from .oauth2 import OAuth2PasswordBearer as OAuth2PasswordBearer
+from .oauth2 import OAuth2PasswordRequestForm as OAuth2PasswordRequestForm
+from .oauth2 import OAuth2PasswordRequestFormStrict as OAuth2PasswordRequestFormStrict
+from .oauth2 import SecurityScopes as SecurityScopes
+from .open_id_connect_url import OpenIdConnect as OpenIdConnect
@@ -0,0 +1,92 @@
+from typing import Optional
+
+from fastapi.openapi.models import APIKey, APIKeyIn
+from fastapi.security.base import SecurityBase
+from starlette.exceptions import HTTPException
+from starlette.requests import Request
+from starlette.status import HTTP_403_FORBIDDEN
+
+
+class APIKeyBase(SecurityBase):
+    pass
+
+
+class APIKeyQuery(APIKeyBase):
+    def __init__(
+        self,
+        *,
+        name: str,
+        scheme_name: Optional[str] = None,
+        description: Optional[str] = None,
+        auto_error: bool = True,
+    ):
+        self.model: APIKey = APIKey(
+            **{"in": APIKeyIn.query}, name=name, description=description
+        )
+        self.scheme_name = scheme_name or self.__class__.__name__
+        self.auto_error = auto_error
+
+    async def __call__(self, request: Request) -> Optional[str]:
+        api_key = request.query_params.get(self.model.name)
+        if not api_key:
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
+                )
+            else:
+                return None
+        return api_key
+
+
+class APIKeyHeader(APIKeyBase):
+    def __init__(
+        self,
+        *,
+        name: str,
+        scheme_name: Optional[str] = None,
+        description: Optional[str] = None,
+        auto_error: bool = True,
+    ):
+        self.model: APIKey = APIKey(
+            **{"in": APIKeyIn.header}, name=name, description=description
+        )
+        self.scheme_name = scheme_name or self.__class__.__name__
+        self.auto_error = auto_error
+
+    async def __call__(self, request: Request) -> Optional[str]:
+        api_key = request.headers.get(self.model.name)
+        if not api_key:
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
+                )
+            else:
+                return None
+        return api_key
+
+
+class APIKeyCookie(APIKeyBase):
+    def __init__(
+        self,
+        *,
+        name: str,
+        scheme_name: Optional[str] = None,
+        description: Optional[str] = None,
+        auto_error: bool = True,
+    ):
+        self.model: APIKey = APIKey(
+            **{"in": APIKeyIn.cookie}, name=name, description=description
+        )
+        self.scheme_name = scheme_name or self.__class__.__name__
+        self.auto_error = auto_error
+
+    async def __call__(self, request: Request) -> Optional[str]:
+        api_key = request.cookies.get(self.model.name)
+        if not api_key:
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
+                )
+            else:
+                return None
+        return api_key
@@ -0,0 +1,6 @@
+from fastapi.openapi.models import SecurityBase as SecurityBaseModel
+
+
+class SecurityBase:
+    model: SecurityBaseModel
+    scheme_name: str
@@ -0,0 +1,165 @@
+import binascii
+from base64 import b64decode
+from typing import Optional
+
+from fastapi.exceptions import HTTPException
+from fastapi.openapi.models import HTTPBase as HTTPBaseModel
+from fastapi.openapi.models import HTTPBearer as HTTPBearerModel
+from fastapi.security.base import SecurityBase
+from fastapi.security.utils import get_authorization_scheme_param
+from pydantic import BaseModel
+from starlette.requests import Request
+from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
+
+
+class HTTPBasicCredentials(BaseModel):
+    username: str
+    password: str
+
+
+class HTTPAuthorizationCredentials(BaseModel):
+    scheme: str
+    credentials: str
+
+
+class HTTPBase(SecurityBase):
+    def __init__(
+        self,
+        *,
+        scheme: str,
+        scheme_name: Optional[str] = None,
+        description: Optional[str] = None,
+        auto_error: bool = True,
+    ):
+        self.model = HTTPBaseModel(scheme=scheme, description=description)
+        self.scheme_name = scheme_name or self.__class__.__name__
+        self.auto_error = auto_error
+
+    async def __call__(
+        self, request: Request
+    ) -> Optional[HTTPAuthorizationCredentials]:
+        authorization = request.headers.get("Authorization")
+        scheme, credentials = get_authorization_scheme_param(authorization)
+        if not (authorization and scheme and credentials):
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
+                )
+            else:
+                return None
+        return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
+
+
+class HTTPBasic(HTTPBase):
+    def __init__(
+        self,
+        *,
+        scheme_name: Optional[str] = None,
+        realm: Optional[str] = None,
+        description: Optional[str] = None,
+        auto_error: bool = True,
+    ):
+        self.model = HTTPBaseModel(scheme="basic", description=description)
+        self.scheme_name = scheme_name or self.__class__.__name__
+        self.realm = realm
+        self.auto_error = auto_error
+
+    async def __call__(  # type: ignore
+        self, request: Request
+    ) -> Optional[HTTPBasicCredentials]:
+        authorization = request.headers.get("Authorization")
+        scheme, param = get_authorization_scheme_param(authorization)
+        if self.realm:
+            unauthorized_headers = {"WWW-Authenticate": f'Basic realm="{self.realm}"'}
+        else:
+            unauthorized_headers = {"WWW-Authenticate": "Basic"}
+        invalid_user_credentials_exc = HTTPException(
+            status_code=HTTP_401_UNAUTHORIZED,
+            detail="Invalid authentication credentials",
+            headers=unauthorized_headers,
+        )
+        if not authorization or scheme.lower() != "basic":
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_401_UNAUTHORIZED,
+                    detail="Not authenticated",
+                    headers=unauthorized_headers,
+                )
+            else:
+                return None
+        try:
+            data = b64decode(param).decode("ascii")
+        except (ValueError, UnicodeDecodeError, binascii.Error):
+            raise invalid_user_credentials_exc
+        username, separator, password = data.partition(":")
+        if not separator:
+            raise invalid_user_credentials_exc
+        return HTTPBasicCredentials(username=username, password=password)
+
+
+class HTTPBearer(HTTPBase):
+    def __init__(
+        self,
+        *,
+        bearerFormat: Optional[str] = None,
+        scheme_name: Optional[str] = None,
+        description: Optional[str] = None,
+        auto_error: bool = True,
+    ):
+        self.model = HTTPBearerModel(bearerFormat=bearerFormat, description=description)
+        self.scheme_name = scheme_name or self.__class__.__name__
+        self.auto_error = auto_error
+
+    async def __call__(
+        self, request: Request
+    ) -> Optional[HTTPAuthorizationCredentials]:
+        authorization = request.headers.get("Authorization")
+        scheme, credentials = get_authorization_scheme_param(authorization)
+        if not (authorization and scheme and credentials):
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
+                )
+            else:
+                return None
+        if scheme.lower() != "bearer":
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_403_FORBIDDEN,
+                    detail="Invalid authentication credentials",
+                )
+            else:
+                return None
+        return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
+
+
+class HTTPDigest(HTTPBase):
+    def __init__(
+        self,
+        *,
+        scheme_name: Optional[str] = None,
+        description: Optional[str] = None,
+        auto_error: bool = True,
+    ):
+        self.model = HTTPBaseModel(scheme="digest", description=description)
+        self.scheme_name = scheme_name or self.__class__.__name__
+        self.auto_error = auto_error
+
+    async def __call__(
+        self, request: Request
+    ) -> Optional[HTTPAuthorizationCredentials]:
+        authorization = request.headers.get("Authorization")
+        scheme, credentials = get_authorization_scheme_param(authorization)
+        if not (authorization and scheme and credentials):
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
+                )
+            else:
+                return None
+        if scheme.lower() != "digest":
+            raise HTTPException(
+                status_code=HTTP_403_FORBIDDEN,
+                detail="Invalid authentication credentials",
+            )
+        return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
@@ -0,0 +1,220 @@
+from typing import Any, Dict, List, Optional, Union
+
+from fastapi.exceptions import HTTPException
+from fastapi.openapi.models import OAuth2 as OAuth2Model
+from fastapi.openapi.models import OAuthFlows as OAuthFlowsModel
+from fastapi.param_functions import Form
+from fastapi.security.base import SecurityBase
+from fastapi.security.utils import get_authorization_scheme_param
+from starlette.requests import Request
+from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
+
+
+class OAuth2PasswordRequestForm:
+    """
+    This is a dependency class, use it like:
+
+        @app.post("/login")
+        def login(form_data: OAuth2PasswordRequestForm = Depends()):
+            data = form_data.parse()
+            print(data.username)
+            print(data.password)
+            for scope in data.scopes:
+                print(scope)
+            if data.client_id:
+                print(data.client_id)
+            if data.client_secret:
+                print(data.client_secret)
+            return data
+
+
+    It creates the following Form request parameters in your endpoint:
+
+    grant_type: the OAuth2 spec says it is required and MUST be the fixed string "password".
+        Nevertheless, this dependency class is permissive and allows not passing it. If you want to enforce it,
+        use instead the OAuth2PasswordRequestFormStrict dependency.
+    username: username string. The OAuth2 spec requires the exact field name "username".
+    password: password string. The OAuth2 spec requires the exact field name "password".
+    scope: Optional string. Several scopes (each one a string) separated by spaces. E.g.
+        "items:read items:write users:read profile openid"
+    client_id: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
+        using HTTP Basic auth, as: client_id:client_secret
+    client_secret: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
+        using HTTP Basic auth, as: client_id:client_secret
+    """
+
+    def __init__(
+        self,
+        grant_type: str = Form(default=None, regex="password"),
+        username: str = Form(),
+        password: str = Form(),
+        scope: str = Form(default=""),
+        client_id: Optional[str] = Form(default=None),
+        client_secret: Optional[str] = Form(default=None),
+    ):
+        self.grant_type = grant_type
+        self.username = username
+        self.password = password
+        self.scopes = scope.split()
+        self.client_id = client_id
+        self.client_secret = client_secret
+
+
+class OAuth2PasswordRequestFormStrict(OAuth2PasswordRequestForm):
+    """
+    This is a dependency class, use it like:
+
+        @app.post("/login")
+        def login(form_data: OAuth2PasswordRequestFormStrict = Depends()):
+            data = form_data.parse()
+            print(data.username)
+            print(data.password)
+            for scope in data.scopes:
+                print(scope)
+            if data.client_id:
+                print(data.client_id)
+            if data.client_secret:
+                print(data.client_secret)
+            return data
+
+
+    It creates the following Form request parameters in your endpoint:
+
+    grant_type: the OAuth2 spec says it is required and MUST be the fixed string "password".
+        This dependency is strict about it. If you want to be permissive, use instead the
+        OAuth2PasswordRequestForm dependency class.
+    username: username string. The OAuth2 spec requires the exact field name "username".
+    password: password string. The OAuth2 spec requires the exact field name "password".
+    scope: Optional string. Several scopes (each one a string) separated by spaces. E.g.
+        "items:read items:write users:read profile openid"
+    client_id: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
+        using HTTP Basic auth, as: client_id:client_secret
+    client_secret: optional string. OAuth2 recommends sending the client_id and client_secret (if any)
+        using HTTP Basic auth, as: client_id:client_secret
+    """
+
+    def __init__(
+        self,
+        grant_type: str = Form(regex="password"),
+        username: str = Form(),
+        password: str = Form(),
+        scope: str = Form(default=""),
+        client_id: Optional[str] = Form(default=None),
+        client_secret: Optional[str] = Form(default=None),
+    ):
+        super().__init__(
+            grant_type=grant_type,
+            username=username,
+            password=password,
+            scope=scope,
+            client_id=client_id,
+            client_secret=client_secret,
+        )
+
+
+class OAuth2(SecurityBase):
+    def __init__(
+        self,
+        *,
+        flows: Union[OAuthFlowsModel, Dict[str, Dict[str, Any]]] = OAuthFlowsModel(),
+        scheme_name: Optional[str] = None,
+        description: Optional[str] = None,
+        auto_error: bool = True,
+    ):
+        self.model = OAuth2Model(flows=flows, description=description)
+        self.scheme_name = scheme_name or self.__class__.__name__
+        self.auto_error = auto_error
+
+    async def __call__(self, request: Request) -> Optional[str]:
+        authorization = request.headers.get("Authorization")
+        if not authorization:
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
+                )
+            else:
+                return None
+        return authorization
+
+
+class OAuth2PasswordBearer(OAuth2):
+    def __init__(
+        self,
+        tokenUrl: str,
+        scheme_name: Optional[str] = None,
+        scopes: Optional[Dict[str, str]] = None,
+        description: Optional[str] = None,
+        auto_error: bool = True,
+    ):
+        if not scopes:
+            scopes = {}
+        flows = OAuthFlowsModel(password={"tokenUrl": tokenUrl, "scopes": scopes})
+        super().__init__(
+            flows=flows,
+            scheme_name=scheme_name,
+            description=description,
+            auto_error=auto_error,
+        )
+
+    async def __call__(self, request: Request) -> Optional[str]:
+        authorization = request.headers.get("Authorization")
+        scheme, param = get_authorization_scheme_param(authorization)
+        if not authorization or scheme.lower() != "bearer":
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_401_UNAUTHORIZED,
+                    detail="Not authenticated",
+                    headers={"WWW-Authenticate": "Bearer"},
+                )
+            else:
+                return None
+        return param
+
+
+class OAuth2AuthorizationCodeBearer(OAuth2):
+    def __init__(
+        self,
+        authorizationUrl: str,
+        tokenUrl: str,
+        refreshUrl: Optional[str] = None,
+        scheme_name: Optional[str] = None,
+        scopes: Optional[Dict[str, str]] = None,
+        description: Optional[str] = None,
+        auto_error: bool = True,
+    ):
+        if not scopes:
+            scopes = {}
+        flows = OAuthFlowsModel(
+            authorizationCode={
+                "authorizationUrl": authorizationUrl,
+                "tokenUrl": tokenUrl,
+                "refreshUrl": refreshUrl,
+                "scopes": scopes,
+            }
+        )
+        super().__init__(
+            flows=flows,
+            scheme_name=scheme_name,
+            description=description,
+            auto_error=auto_error,
+        )
+
+    async def __call__(self, request: Request) -> Optional[str]:
+        authorization = request.headers.get("Authorization")
+        scheme, param = get_authorization_scheme_param(authorization)
+        if not authorization or scheme.lower() != "bearer":
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_401_UNAUTHORIZED,
+                    detail="Not authenticated",
+                    headers={"WWW-Authenticate": "Bearer"},
+                )
+            else:
+                return None  # pragma: nocover
+        return param
+
+
+class SecurityScopes:
+    def __init__(self, scopes: Optional[List[str]] = None):
+        self.scopes = scopes or []
+        self.scope_str = " ".join(self.scopes)
@@ -0,0 +1,34 @@
+from typing import Optional
+
+from fastapi.openapi.models import OpenIdConnect as OpenIdConnectModel
+from fastapi.security.base import SecurityBase
+from starlette.exceptions import HTTPException
+from starlette.requests import Request
+from starlette.status import HTTP_403_FORBIDDEN
+
+
+class OpenIdConnect(SecurityBase):
+    def __init__(
+        self,
+        *,
+        openIdConnectUrl: str,
+        scheme_name: Optional[str] = None,
+        description: Optional[str] = None,
+        auto_error: bool = True,
+    ):
+        self.model = OpenIdConnectModel(
+            openIdConnectUrl=openIdConnectUrl, description=description
+        )
+        self.scheme_name = scheme_name or self.__class__.__name__
+        self.auto_error = auto_error
+
+    async def __call__(self, request: Request) -> Optional[str]:
+        authorization = request.headers.get("Authorization")
+        if not authorization:
+            if self.auto_error:
+                raise HTTPException(
+                    status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
+                )
+            else:
+                return None
+        return authorization
@@ -0,0 +1,10 @@
+from typing import Optional, Tuple
+
+
+def get_authorization_scheme_param(
+    authorization_header_value: Optional[str],
+) -> Tuple[str, str]:
+    if not authorization_header_value:
+        return "", ""
+    scheme, _, param = authorization_header_value.partition(" ")
+    return scheme, param