diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index b68511b7d43b2..1b201005ba7fd 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -315,6 +315,8 @@ repos: pass_filenames: true exclude: > (?x) + ^airflow/api_connexion/endpoints/auth_endpoint.py$| + ^airflow/api_connexion/security.py$| ^airflow/providers/apache/cassandra/hooks/cassandra.py$| ^airflow/providers/apache/hive/operators/hive_stats.py$| ^airflow/providers/apache/hive/.*PROVIDER_CHANGES_*| diff --git a/airflow/api_connexion/endpoints/auth_endpoint.py b/airflow/api_connexion/endpoints/auth_endpoint.py new file mode 100644 index 0000000000000..f4515df0ad4b0 --- /dev/null +++ b/airflow/api_connexion/endpoints/auth_endpoint.py @@ -0,0 +1,242 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +import logging +import re + +import jwt +from flask import current_app, jsonify, request +from flask_appbuilder.const import AUTH_DB, AUTH_LDAP, AUTH_OAUTH, AUTH_OID, AUTH_REMOTE_USER +from flask_jwt_extended import ( + create_access_token, + decode_token, + get_jwt_identity, + get_raw_jwt, + jwt_refresh_token_required, + jwt_required, +) +from flask_login import login_user +from marshmallow import ValidationError +from sqlalchemy.exc import IntegrityError + +from airflow.api_connexion.exceptions import BadRequest, Unauthenticated +from airflow.api_connexion.schemas.auth_schema import ( + auth_schema, + info_schema, + login_form_schema, + token_schema, +) +from airflow.configuration import conf +from airflow.models.auth import TokenBlockList + +log = logging.getLogger(__name__) + + +def get_auth_info(): + """Get site authentication info""" + security_manager = current_app.appbuilder.sm + config = current_app.config + auth_type = security_manager.auth_type + type_mapping = { + AUTH_DB: "auth_db", + AUTH_LDAP: "auth_ldap", + AUTH_OID: "auth_oid", + AUTH_OAUTH: "auth_oauth", + AUTH_REMOTE_USER: "auth_remote_user", + } + oauth_providers = config.get("OAUTH_PROVIDERS", None) + openid_providers = config.get("OPENID_PROVIDERS", None) + return info_schema.dump( + { + "auth_type": type_mapping[auth_type], + "oauth_providers": oauth_providers, + "openid_providers": openid_providers, + } + ) + + +def auth_login(): + """Handle DB login""" + appbuilder = current_app.appbuilder + body = request.json + try: + data = login_form_schema.load(body) + except ValidationError as err: + raise BadRequest(detail=str(err.messages)) + security_manager = appbuilder.sm + if security_manager.auth_type not in [AUTH_DB, AUTH_LDAP]: + raise Unauthenticated(detail="Authentication type does not match") + user = security_manager.login_with_user_pass(data['username'], data['password']) + if not user: + raise Unauthenticated(detail="Invalid login") + login_user(user, remember=False) + return security_manager.create_tokens_and_dump(user, auth_schema) + + +def auth_oauthlogin(provider, register=None): + """Authenticates user and redirect to callback endpoint""" + appbuilder = current_app.appbuilder + security_manager = appbuilder.sm + if security_manager.auth_type != AUTH_OAUTH: + raise Unauthenticated(detail="Authentication type does not match") + api_base_path = "/api/v1/" + base_url = conf.get("webserver", "base_url") + api_base_path + redirect_uri = base_url + "oauth-authorized?provider=" + provider + + arg = {'register': [str(register)]} if register else {} + state = jwt.encode( + arg, + key=appbuilder.app.config['SECRET_KEY'], + algorithm='HS256', + ) + auth_provider = security_manager.oauth_remotes[provider] + try: + if provider == "twitter": + redirect_uri = redirect_uri + f"&state={state}" + return auth_provider.authorize_redirect(redirect_uri=redirect_uri) + else: + state = state.decode("ascii") if isinstance(state, bytes) else state + return auth_provider.authorize_redirect( + redirect_uri=redirect_uri, + state=state, + ) + except Exception as err: # pylint: disable=broad-except + log.error("Error on OAuth authorize: %s", err) + raise Unauthenticated(detail=str(err)) + + +def authorize_oauth(): + """Callback to authorize Oauth.""" + args = request.args + provider = args.get('provider', None) + state = args.get('state', None) + if not (provider and state): + raise BadRequest(detail="Missing required parameters: provider and state") + appbuilder = current_app.appbuilder + security_manager = appbuilder.sm + log.debug("Authorized init") + resp = security_manager.oauth_remotes[provider].authorize_access_token() + if resp is None: + raise Unauthenticated(detail="You denied the request to sign in") + # Verify state + try: + jwt.decode( + state, + key=appbuilder.app.config['SECRET_KEY'], + algorithms='HS256', + ) + except jwt.InvalidTokenError: + raise BadRequest(detail="State signature is not valid!") + log.debug("OAUTH Authorized resp: %s", resp) + # Retrieves specific user info from the provider + try: # pylint: disable=too-many-nested-blocks + userinfo = security_manager.oauth_user_info(provider, resp) + except Exception as err: # pylint: disable=broad-except + log.debug("Error retrieving user info. Error =>%s", err) + user = None + else: + log.debug("User info retrieved from %s: %s", provider, userinfo) + # User email is not whitelisted + if provider in security_manager.oauth_whitelists: + whitelist = security_manager.oauth_whitelists[provider] + allow = False + for e in whitelist: + if re.search(e, userinfo["email"]): + allow = True + break + if not allow: + raise Unauthenticated(detail="You are not authorized.") + else: + log.debug("No allow list for OAuth provider") + user = security_manager.auth_user_oauth(userinfo) + log.debug("User is %s", user) + if user is None: + raise Unauthenticated(detail="Invalid login") + login_user(user, remember=False) + return security_manager.create_tokens_and_dump(user, auth_schema) + + +def auth_remoteuser(): + """Handle remote user auth""" + appbuilder = current_app.appbuilder + security_manager = appbuilder.sm + if security_manager.auth_type != AUTH_REMOTE_USER: + raise Unauthenticated(detail="Authentication type does not match") + username = request.environ.get("REMOTE_USER") + if not username: + raise Unauthenticated(detail="Invalid login") + user = security_manager.auth_user_remote_user(username) + if user is None: + raise Unauthenticated(detail="Invalid login") + login_user(user, remember=False) + return security_manager.create_tokens_and_dump(user, auth_schema) + + +@jwt_refresh_token_required +def refresh_token(): + """Refresh token""" + user = get_jwt_identity() + access_token = create_access_token(identity=user) + return {'access_token': access_token} + + +@jwt_required +def revoke_token(): + """ + An endpoint for revoking both access and refresh token. + + This is intended for a case where a logged in user want to revoke + another user's tokens + """ + body = request.json + try: + data = token_schema.load(body) + except ValidationError as err: + raise BadRequest(detail=str(err.messages)) + token = decode_token(data['token']) + try: + TokenBlockList.add_token(jti=token['jti'], expiry_delta=token['exp']) + except IntegrityError: + pass + return jsonify({"revoked": True}) + + +@jwt_required +def revoke_current_user_token(): + """ + An endpoint for revoking current user access token + This should be called during current user logout + """ + raw_jwt = get_raw_jwt() + try: + TokenBlockList.add_token(jti=raw_jwt['jti'], expiry_delta=raw_jwt['exp']) + except IntegrityError: + pass + return jsonify({"revoked": True}) + + +@jwt_refresh_token_required +def revoke_current_user_refresh_token(): + """ + An endpoint for revoking current user refresh token + This should be called during current user logout + """ + raw_jwt = get_raw_jwt() + try: + TokenBlockList.add_token(jti=raw_jwt['jti'], expiry_delta=raw_jwt['exp']) + except IntegrityError: + pass + return jsonify({"revoked": True}) diff --git a/airflow/api_connexion/openapi/v1.yaml b/airflow/api_connexion/openapi/v1.yaml index 2fd58ea326955..8d8e25b09698d 100644 --- a/airflow/api_connexion/openapi/v1.yaml +++ b/airflow/api_connexion/openapi/v1.yaml @@ -1572,6 +1572,192 @@ paths: '404': $ref: '#/components/responses/NotFound' + /auth-info: + get: + summary: Get site authentication information + x-openapi-router-controller: airflow.api_connexion.endpoints.auth_endpoint + operationId: get_auth_info + tags: [Authentication] + responses: + '200': + description: Success. + content: + application/json: + schema: + $ref: '#/components/schemas/AuthInfo' + + /auth/login: + post: + summary: Login user + x-openapi-router-controller: airflow.api_connexion.endpoints.auth_endpoint + operationId: auth_login + tags: [Authentication] + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/AuthDBLoginForm' + responses: + '200': + description: Success. + content: + application/json: + schema: + $ref: '#/components/schemas/JwtAuthAndUser' + '401': + $ref: '#/components/responses/Unauthenticated' + + + /auth-oauth/{provider}: + get: + summary: OAUTH user login + x-openapi-router-controller: airflow.api_connexion.endpoints.auth_endpoint + operationId: auth_oauthlogin + tags: [Authentication] + parameters: + - $ref: '#/components/parameters/ShouldRegister' + - $ref: '#/components/parameters/OauthProvider' + responses: + '302': + description: Redirect. + headers: + Location: + schema: + type: string + format: uri + + '401': + $ref: '#/components/responses/Unauthenticated' + + /oauth-authorized: + get: + summary: Oauth authentication + x-openapi-router-controller: airflow.api_connexion.endpoints.auth_endpoint + operationId: authorize_oauth + tags: [Authentication] + responses: + '200': + description: Success. + content: + application/json: + schema: + $ref: '#/components/schemas/JwtAuthAndUser' + '401': + $ref: '#/components/responses/Unauthenticated' + + + /auth-remoteuser: + get: + summary: Authenticate using remote user auth + x-openapi-router-controller: airflow.api_connexion.endpoints.auth_endpoint + operationId: auth_remoteuser + tags: [Authentication] + responses: + '200': + description: Success. + content: + application/json: + schema: + $ref: '#/components/schemas/JwtAuthAndUser' + '401': + $ref: '#/components/responses/Unauthenticated' + + /refresh: + get: + summary: Refresh a token + description: Accepts refresh token and creates access token + x-openapi-router-controller: airflow.api_connexion.endpoints.auth_endpoint + operationId: refresh_token + tags: [Authentication] + responses: + '200': + description: Success + content: + application/json: + schema: + $ref: '#/components/schemas/AccessToken' + '401': + $ref: '#/components/responses/Unauthenticated' + + /revoke_access_token: + get: + summary: Revoke current user access token + description: | + An endpoint for revoking current user access token + This should be called during current user logout + x-openapi-router-controller: airflow.api_connexion.endpoints.auth_endpoint + operationId: revoke_current_user_token + tags: [Authentication] + responses: + '200': + description: Success + content: + application/json: + schema: + type: object + properties: + revoked: + type: boolean + description: Indicate user is logged out + '401': + $ref: '#/components/responses/Unauthenticated' + + /revoke_refresh_token: + get: + summary: Revoke current user refresh token + description: | + An endpoint for revoking current user access token + This should be called during current user logout + x-openapi-router-controller: airflow.api_connexion.endpoints.auth_endpoint + operationId: revoke_current_user_refresh_token + tags: [Authentication] + responses: + '200': + description: Success + content: + application/json: + schema: + type: object + properties: + revoked: + type: boolean + description: Indicate user is logged out + '401': + $ref: '#/components/responses/Unauthenticated' + + /revoke: + post: + summary: Revoke access and refresh token + description: | + An endpoint for revoking both access and refresh token. + x-openapi-router-controller: airflow.api_connexion.endpoints.auth_endpoint + operationId: revoke_token + tags: [Authentication] + requestBody: + required: true + content: + application/json: + schema: + $ref: '#/components/schemas/Revoke' + responses: + '200': + description: Success + content: + application/json: + schema: + type: object + properties: + revoked: + type: boolean + description: Indicate token is revoked + '404': + $ref: '#/components/responses/NotFound' + '401': + $ref: '#/components/responses/Unauthenticated' + '400': + $ref: '#/components/responses/BadRequest' + components: # Reusable schemas (data models) schemas: @@ -2508,6 +2694,73 @@ components: type: object $ref: '#/components/schemas/Resource' description: The permission resource + # Auth + AuthInfo: + type: object + description: Information about the authentication method in use + properties: + auth_type: + type: string + description: Authentication type in use + oauth_providers: + type: array + items: + type: object + nullable: true + openid_providers: + type: array + items: + type: object + nullable: true + + AuthDBLoginForm: + type: object + properties: + username: + description: The username for authentication + example: admin + type: string + password: + description: The password for authentication + example: complex-password + type: string + + AuthorizationUrl: + description: Oauth authorization URl + type: object + properties: + auth_url: + type: string + + JwtAuthAndUser: + description: Data returned after user authentication + type: object + allOf: + - type: object + properties: + token: + type: string + description: Authentication token + refresh_token: + type: string + description: Authentication refresh token + - $ref: '#/components/schemas/UserCollectionItem' + + AccessToken: + description: Access Token + type: object + properties: + access_token: + type: string + description: authentication token + + Revoke: + description: Revoke a token + type: object + properties: + token: + type: string + description: The token to be revoked # Configuration ConfigOption: @@ -2555,6 +2808,16 @@ components: nullable: true # Form + AccessRefresh: + type: object + properties: + token: + description: Access token + type: string + refresh_token: + description: Refresh token + type: string + ClearTaskInstance: type: object properties: @@ -3266,8 +3529,24 @@ components: The name of the field to order the results by. Prefix a field name with `-` to reverse the sort order. - # Other parameters + # Auth parameters + OauthProvider: + in: path + name: provider + schema: + type: string + required: true + description: The name of the provider to authenticate + + ShouldRegister: + in: query + name: register + schema: + type: string + required: false + description: Whether to register user + # Other parameters FileToken: in: path name: file_token @@ -3376,6 +3655,10 @@ components: Kerberos: type: http scheme: negotiate + bearerAuth: + type: http + scheme: bearer + bearerFormat: JWT # The API will provide support for plugins to support various authorization mechanisms. # Detailed information will be available in the plugin specification. @@ -3397,6 +3680,7 @@ tags: - name: Role - name: Permission - name: User + - name: Authentication externalDocs: url: https://airflow.apache.org/docs/apache-airflow/stable/ diff --git a/airflow/api_connexion/schemas/auth_schema.py b/airflow/api_connexion/schemas/auth_schema.py new file mode 100644 index 0000000000000..e7f82da14df5f --- /dev/null +++ b/airflow/api_connexion/schemas/auth_schema.py @@ -0,0 +1,70 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from marshmallow import fields +from marshmallow.schema import Schema + +from airflow.api_connexion.schemas.user_schema import UserCollectionItemSchema + + +class OAUTHSchema(Schema): + """A little information needed for UI""" + + name = fields.String() + icon = fields.String() + + +class OpenIDSchema(Schema): + """A little information needed for the UI""" + + name = fields.String() + url = fields.String() + + +class InfoSchema(Schema): + """Authentication Info Schema""" + + auth_type = fields.String() + oauth_providers = fields.List(fields.Nested(OAUTHSchema)) + openid_providers = fields.List(fields.Nested(OpenIDSchema)) + + +class LoginForm(Schema): + """Used to load credentials""" + + username = fields.String(required=True) + password = fields.String(required=True) + + +class AuthSchema(Schema): + """Schema for data returned after successful authentication""" + + token = fields.String() + refresh_token = fields.String() + user = fields.Nested(UserCollectionItemSchema) + + +class RevokeTokenSchema(Schema): + """Schema to revoke tokens""" + + token = fields.String(required=True) + + +info_schema = InfoSchema() +login_form_schema = LoginForm() +auth_schema = AuthSchema() +token_schema = RevokeTokenSchema() diff --git a/airflow/api_connexion/security.py b/airflow/api_connexion/security.py index 4faa9ed8f3457..86f209b1e038b 100644 --- a/airflow/api_connexion/security.py +++ b/airflow/api_connexion/security.py @@ -14,24 +14,38 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. - from functools import wraps from typing import Callable, Optional, Sequence, Tuple, TypeVar, cast from flask import Response, current_app +from flask_jwt_extended import verify_jwt_in_request from airflow.api_connexion.exceptions import PermissionDenied, Unauthenticated +from airflow.models import TokenBlockList T = TypeVar("T", bound=Callable) # pylint: disable=invalid-name +@current_app.appbuilder.sm.jwt_manager.token_in_blacklist_loader +def check_if_token_in_blacklist(decrypted_token): + """Checks if there's a blocked token""" + jti = decrypted_token['jti'] + return TokenBlockList.get_token(jti) is not None + + def check_authentication() -> None: """Checks that the request has valid authorization information.""" response = current_app.api_auth.requires_authentication(Response)() - if response.status_code != 200: + if response.status_code == 200: + return + try: + verify_jwt_in_request() + return + except Exception: # pylint: disable=broad-except + pass # since this handler only checks authentication, not authorization, # we should always return 401 - raise Unauthenticated(headers=response.headers) + raise Unauthenticated(headers=response.headers) def requires_access(permissions: Optional[Sequence[Tuple[str, str]]] = None) -> Callable[[T], T]: diff --git a/airflow/config_templates/default_webserver_config.py b/airflow/config_templates/default_webserver_config.py index 7f19ff129bd4d..7605438a77326 100644 --- a/airflow/config_templates/default_webserver_config.py +++ b/airflow/config_templates/default_webserver_config.py @@ -17,6 +17,7 @@ # under the License. """Default configuration for the Airflow webserver""" import os +from datetime import timedelta from flask_appbuilder.security.manager import AUTH_DB @@ -126,3 +127,15 @@ # APP_THEME = "superhero.css" # APP_THEME = "united.css" # APP_THEME = "yeti.css" +# ---------------------------------------------------- +# JWT CONFIGURATION +# ---------------------------------------------------- +# Airflow uses flask_jwt_extended for handling JWT Authentication. +# Please refer to https://flask-jwt-extended.readthedocs.io/en/3.0.0_release/options/ +# for more configuration options. +# Note that storing token in query_string or cookies is not supported +JWT_TOKEN_LOCATION = ['headers'] +JWT_HEADER_NAME = "Authorization" +JWT_HEADER_TYPE = 'Bearer' +JWT_ACCESS_TOKEN_EXPIRES = timedelta(minutes=30) +JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=30) diff --git a/airflow/migrations/versions/22ab4efd5674_add_token_blocklist.py b/airflow/migrations/versions/22ab4efd5674_add_token_blocklist.py new file mode 100644 index 0000000000000..9d190a9e39214 --- /dev/null +++ b/airflow/migrations/versions/22ab4efd5674_add_token_blocklist.py @@ -0,0 +1,53 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +"""add-token-blocklist + +Revision ID: 22ab4efd5674 +Revises: a13f7613ad25 +Create Date: 2021-04-17 18:16:31.019394 + +""" + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision = '22ab4efd5674' +down_revision = 'a13f7613ad25' +branch_labels = None +depends_on = None + +INDEX_NAME = "idx_expiry_date_token_blocklist" +TABLE_NAME = "token_blocklist" + + +def upgrade(): + """Apply Add token blocklist table""" + op.create_table( + TABLE_NAME, + sa.Column("jti", sa.String(50), nullable=False, primary_key=True), + sa.Column("expiry_date", sa.DateTime(), nullable=False), + ) + op.create_index(INDEX_NAME, TABLE_NAME, ['expiry_date'], unique=False) + + +def downgrade(): # noqa: D103 + """Unapply Add token blocklist table""" + op.drop_index(INDEX_NAME, table_name=TABLE_NAME) + op.drop_table(TABLE_NAME) diff --git a/airflow/models/__init__.py b/airflow/models/__init__.py index 61606ea86b601..e6650e03ffa85 100644 --- a/airflow/models/__init__.py +++ b/airflow/models/__init__.py @@ -16,6 +16,7 @@ # specific language governing permissions and limitations # under the License. """Airflow models""" +from airflow.models.auth import TokenBlockList from airflow.models.base import ID_LEN, Base from airflow.models.baseoperator import BaseOperator, BaseOperatorLink from airflow.models.connection import Connection diff --git a/airflow/models/auth.py b/airflow/models/auth.py new file mode 100644 index 0000000000000..130ddaa5504f3 --- /dev/null +++ b/airflow/models/auth.py @@ -0,0 +1,68 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from datetime import datetime + +from pendulum import from_timestamp +from sqlalchemy import Column, DateTime, Index, String + +from airflow.models.base import Base +from airflow.utils.log.logging_mixin import LoggingMixin +from airflow.utils.session import provide_session + + +class TokenBlockList(Base, LoggingMixin): + """ + A model for recording blocked token + + :param jti: The token jti(JWT ID) + :type jti: str + :param expiry_date: When the token would expire + :type expiry_date: DateTime + """ + + __tablename__ = 'token_blocklist' + jti = Column(String(50), nullable=False, primary_key=True) + expiry_date = Column(DateTime(), nullable=False) + __table_args__ = (Index('idx_expiry_date_token_blocklist', expiry_date),) + + @classmethod + @provide_session + def get_token(cls, token, session=None): + """Get a token""" + return session.query(cls).get(token) + + @classmethod + @provide_session + def delete_token(cls, token, session=None): + """Delete a token""" + session.query(cls).filter(cls.jti == token).delete() + + @classmethod + @provide_session + def add_token(cls, jti, expiry_delta, session=None): + """Add a token to blocklist""" + # pylint: disable=unexpected-keyword-arg + token = cls(jti=jti, expiry_date=from_timestamp(expiry_delta)) + session.add(token) + session.commit() + + @classmethod + @provide_session + def delete_expired_tokens(cls, session=None): + """Delete all expired tokens""" + session.query(cls).filter(cls.expiry_date <= datetime.now()).delete() diff --git a/airflow/www/app.py b/airflow/www/app.py index 632e1d2727c6d..99f1dbad6a318 100644 --- a/airflow/www/app.py +++ b/airflow/www/app.py @@ -129,7 +129,6 @@ def create_app(config=None, testing=False): init_error_handlers(flask_app) init_api_connexion(flask_app) init_api_experimental(flask_app) - sync_appbuilder_roles(flask_app) init_jinja_globals(flask_app) diff --git a/airflow/www/extensions/init_views.py b/airflow/www/extensions/init_views.py index 14e532b7ca9be..6226bdc1d118a 100644 --- a/airflow/www/extensions/init_views.py +++ b/airflow/www/extensions/init_views.py @@ -25,6 +25,7 @@ from airflow.api_connexion.exceptions import common_error_handler from airflow.configuration import conf +from airflow.models import TokenBlockList from airflow.security import permissions from airflow.www.views import lazy_add_provider_discovered_options_to_connection_form @@ -156,6 +157,11 @@ def set_cors_headers_on_response(response): return response +def delete_expired_tokens(): + """Function called before first API request""" + TokenBlockList.delete_expired_tokens() + + def init_api_connexion(app: Flask) -> None: """Initialize Stable API""" base_path = '/api/v1' @@ -185,6 +191,7 @@ def _handle_api_error(ex): app.after_request_funcs.setdefault(api_bp.name, []).append(set_cors_headers_on_response) app.register_error_handler(ProblemException, common_error_handler) app.extensions['csrf'].exempt(api_bp) + app.before_first_request(delete_expired_tokens) def init_api_experimental(app): diff --git a/airflow/www/security.py b/airflow/www/security.py index d1bf9bc5565db..e68b38c3b872d 100644 --- a/airflow/www/security.py +++ b/airflow/www/security.py @@ -16,14 +16,16 @@ # specific language governing permissions and limitations # under the License. # - +import logging import warnings from typing import Dict, Optional, Sequence, Set, Tuple from flask import current_app, g +from flask_appbuilder.const import AUTH_DB, AUTH_LDAP from flask_appbuilder.security.sqla import models as sqla_models from flask_appbuilder.security.sqla.manager import SecurityManager from flask_appbuilder.security.sqla.models import PermissionView, Role, User +from flask_jwt_extended import JWTManager, create_access_token, create_refresh_token from sqlalchemy import or_ from sqlalchemy.orm import joinedload @@ -57,6 +59,8 @@ 'Public', } +log = logging.getLogger(__name__) + class AirflowSecurityManager(SecurityManager, LoggingMixin): # pylint: disable=too-many-public-methods """Custom security manager, which introduces a permission model adapted to Airflow""" @@ -730,6 +734,28 @@ def check_authorization( return True + def login_with_user_pass(self, username, password): + """Convenience method for user login through the API""" + user = None + if self.auth_type == AUTH_DB: + user = self.auth_user_db(username, password) + elif self.auth_type == AUTH_LDAP: + user = self.auth_user_ldap(username, password) + return user + + def create_jwt_manager(self, app) -> JWTManager: + """Called by FAB for us when it wants a configured JWT manager""" + jwt_manager = JWTManager() + jwt_manager.init_app(app) + jwt_manager.user_loader_callback_loader(self.load_user_jwt) + return jwt_manager + + def create_tokens_and_dump(self, user, schema): + """Creates access and refresh token, return user data alongside tokens""" + access_token = create_access_token(user.id) + refresh_token = create_refresh_token(user.id) + return schema.dump({'user': user, 'token': access_token, 'refresh_token': refresh_token}) + class ApplessAirflowSecurityManager(AirflowSecurityManager): """Security Manager that doesn't need the whole flask app""" diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt index 464db0cb14e4d..99442e3f17c94 100644 --- a/docs/spelling_wordlist.txt +++ b/docs/spelling_wordlist.txt @@ -194,6 +194,7 @@ InspectContentResponse InspectTemplate Investorise JPype +JWT Jakob Jarek Jdbc @@ -528,6 +529,7 @@ behaviours bigquery bigtable bitshift +blocklist bolkedebruin booktabs boolean @@ -749,6 +751,7 @@ evo exasol execvp exitcode +exp explict exportingmultiple extensibility @@ -858,6 +861,7 @@ hyperparameter iPython iTerm iam +iat idempotence idempotency ie @@ -920,6 +924,7 @@ joygao js json jthomas +jti jupyter jupytercmd kaxil diff --git a/tests/api_connexion/endpoints/test_auth_endpoint.py b/tests/api_connexion/endpoints/test_auth_endpoint.py new file mode 100644 index 0000000000000..afc65cff0068b --- /dev/null +++ b/tests/api_connexion/endpoints/test_auth_endpoint.py @@ -0,0 +1,367 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from unittest import mock + +import pytest +from flask_appbuilder.const import AUTH_DB, AUTH_LDAP, AUTH_OAUTH, AUTH_OID, AUTH_REMOTE_USER +from sqlalchemy import func + +from airflow.models.auth import TokenBlockList +from airflow.security import permissions +from airflow.utils.session import provide_session +from tests.test_utils.api_connexion_utils import create_user, delete_user +from tests.test_utils.config import conf_vars + +AUTH_TYPE_MISMATCH_MESSAGE = "Authentication type does not match" + +OAUTH_PROVIDERS = [ + { + 'name': 'google', + 'token_key': 'access_token', + 'icon': 'fa-google', + 'remote_app': { + 'api_base_url': 'https://www.googleapis.com/oauth2/v2/', + 'client_kwargs': {'scope': 'email profile'}, + 'access_token_url': 'https://accounts.google.com/o/oauth2/token', + 'authorize_url': 'https://accounts.google.com/o/oauth2/auth', + 'request_token_url': None, + 'client_id': "GOOGLE_KEY", + 'client_secret': "GOOGLE_SECRET_KEY", + }, + } +] + +OPENID_PROVIDERS = [ + {'name': 'Yahoo', 'url': 'https://me.yahoo.com'}, + {'name': 'AOL', 'url': 'http://openid.aol.com/'}, +] + + +@pytest.fixture(scope="module") +def configured_app(minimal_app_for_api): + app = minimal_app_for_api + create_user( + app, + username="test", + role_name="Test", + permissions=[ + (permissions.ACTION_CAN_READ, permissions.RESOURCE_POOL), + ], + ) # type: ignore + + yield app + + delete_user(app, username="test") # type: ignore + + +class TestLoginEndpoint: + @pytest.fixture(autouse=True) + def setup_attrs(self, configured_app) -> None: + self.app = configured_app + self.client = self.app.test_client() # type:ignore + self.session = self.app.appbuilder.get_session + + def auth_type(self, auth): + self.app.config['AUTH_TYPE'] = auth + if auth == AUTH_OAUTH: + self.app.config['OAUTH_PROVIDERS'] = OAUTH_PROVIDERS + self.app.config['OPENID_PROVIDERS'] = None + elif auth == AUTH_OID: + self.app.config['OPENID_PROVIDERS'] = OPENID_PROVIDERS + self.app.config['OAUTH_PROVIDERS'] = None + else: + self.app.config['OPENID_PROVIDERS'] = None + self.app.config['OAUTH_PROVIDERS'] = None + + def teardown_method(self): + tokens = self.session.query(TokenBlockList).all() + for token in tokens: + self.session.delete(token) + self.session.commit() + + +class TestAuthInfo(TestLoginEndpoint): + def test_auth_db_info(self): + self.auth_type(AUTH_DB) + response = self.client.get("api/v1/auth-info") + assert response.json == {"auth_type": "auth_db", "oauth_providers": None, "openid_providers": None} + + def test_auth_ldap_info(self): + self.auth_type(AUTH_LDAP) + response = self.client.get("api/v1/auth-info") + assert response.json == {"auth_type": "auth_ldap", "oauth_providers": None, "openid_providers": None} + + def test_auth_oath_info(self): + self.auth_type(AUTH_OAUTH) + response = self.client.get("api/v1/auth-info") + assert response.json == { + 'auth_type': 'auth_oauth', + 'oauth_providers': [ # Only data necessary to build forms are returned + {'icon': 'fa-google', 'name': 'google'} + ], + 'openid_providers': None, + } + + def test_auth_oid_info(self): + self.auth_type(AUTH_OID) + response = self.client.get("api/v1/auth-info") + assert response.json == { + 'auth_type': 'auth_oid', + 'oauth_providers': None, + 'openid_providers': [ + {'name': 'Yahoo', 'url': 'https://me.yahoo.com'}, + {'name': 'AOL', 'url': 'http://openid.aol.com/'}, + ], + } + + +class TestDBLoginEndpoint(TestLoginEndpoint): + def test_user_can_login(self): + self.auth_type(AUTH_DB) + payload = {"username": "test", "password": "test"} + response = self.client.post('api/v1/auth/login', json=payload) + assert response.json['user']['username'] == 'test' + assert isinstance(response.json['token'], str) + assert isinstance(response.json['refresh_token'], str) + + def test_incorrect_username_raises(self): + self.auth_type(AUTH_DB) + payload = {"username": "tests", "password": "test"} + response = self.client.post('api/v1/auth/login', json=payload) + assert response.status_code == 401 + assert response.json['detail'] == 'Invalid login' + + def test_post_body_conforms(self): + self.auth_type(AUTH_DB) + payload = {"username": "tests"} + response = self.client.post('api/v1/auth/login', json=payload) + assert response.status_code == 400 + assert response.json['detail'] == "{'password': ['Missing data for required field.']}" + + def test_auth_type_must_be_db(self): + self.auth_type(AUTH_OAUTH) + payload = {"username": "test", "password": "test"} + response = self.client.post('api/v1/auth/login', json=payload) + assert response.status_code == 401 + assert response.json['detail'] == AUTH_TYPE_MISMATCH_MESSAGE + + +class TestLDAPLoginEndpoint(TestLoginEndpoint): + def test_user_can_login(self): + self.auth_type(AUTH_LDAP) + self.app.appbuilder.sm.auth_user_ldap = mock.Mock() + user = self.app.appbuilder.sm.find_user(username='test') + self.app.appbuilder.sm.auth_user_ldap.return_value = user + payload = {"username": "test", "password": "test"} + response = self.client.post('api/v1/auth/login', json=payload) + assert response.json['user']['username'] == 'test' + assert isinstance(response.json['token'], str) + assert isinstance(response.json['refresh_token'], str) + + def test_incorrect_username_raises(self): + self.auth_type(AUTH_LDAP) + self.app.appbuilder.sm.auth_user_ldap = mock.Mock() + self.app.appbuilder.sm.auth_user_ldap.return_value = None + payload = {"username": "tests", "password": "test"} + response = self.client.post('api/v1/auth/login', json=payload) + assert response.status_code == 401 + assert response.json['detail'] == 'Invalid login' + + def test_post_body_conforms(self): + self.auth_type(AUTH_LDAP) + payload = {"username": "tests"} + response = self.client.post('api/v1/auth/login', json=payload) + assert response.status_code == 400 + assert response.json['detail'] == "{'password': ['Missing data for required field.']}" + + def test_auth_type_must_be_ldap(self): + self.auth_type(AUTH_OAUTH) + payload = {"username": "test", "password": "test"} + response = self.client.post('api/v1/auth/login', json=payload) + assert response.status_code == 401 + assert response.json['detail'] == AUTH_TYPE_MISMATCH_MESSAGE + + +class TestOauthAuthorizationEndpoint(TestLoginEndpoint): + @conf_vars({("webserver", 'base_url'): 'http://localhost:8080'}) + @mock.patch("airflow.api_connexion.endpoints.auth_endpoint.jwt.encode") + def test_can_redirect_for_google(self, mock_jwt_encode): + self.auth_type(AUTH_OAUTH) + mock_jwt_encode.return_value = "state" + mock_google_auth_provider = mock.Mock() + self.app.appbuilder.sm.oauth_remotes = { + "google": mock_google_auth_provider, + } + mock_auth = mock.MagicMock(return_value="test-val") + mock_google_auth_provider.authorize_redirect = mock_auth + redirect_url = "http://localhost:8080/api/v1/oauth-authorized?provider=google" + self.client.get('api/v1/auth-oauth/google?register=True') + mock_auth.assert_called_once_with(redirect_uri=redirect_url, state='state') + + @conf_vars({("webserver", 'base_url'): 'http://localhost:8080'}) + @mock.patch("airflow.api_connexion.endpoints.auth_endpoint.jwt.encode") + def test_can_redirect_for_twitter(self, mock_jwt_encode): + self.auth_type(AUTH_OAUTH) + mock_jwt_encode.return_value = "state" + mock_twitter_auth_provider = mock.Mock() + self.app.appbuilder.sm.oauth_remotes = { + "twitter": mock_twitter_auth_provider, + } + mock_auth = mock.MagicMock(return_value='val') + mock_twitter_auth_provider.authorize_redirect = mock_auth + redirect_url = "http://localhost:8080/api/v1/oauth-authorized?provider=twitter" + self.client.get('api/v1/auth-oauth/twitter?register=True') + mock_auth.assert_called_once_with(redirect_uri=redirect_url + "&state=state") + + def test_incorrect_auth_type_raises(self): + self.auth_type(AUTH_DB) + resp = self.client.get('api/v1/auth-oauth/google?register=True') + assert resp.status_code == 401 + assert resp.json['detail'] == AUTH_TYPE_MISMATCH_MESSAGE + + +class TestAuthorizeOauth(TestLoginEndpoint): + def test_user_refused_sign_in_request(self): + self.auth_type(AUTH_OAUTH) + mock_twitter_auth_provider = mock.Mock() + self.app.appbuilder.sm.oauth_remotes = { + "twitter": mock_twitter_auth_provider, + } + mock_twitter_auth_provider.authorize_access_token.return_value = None + response = self.client.get('api/v1/oauth-authorized?provider=twitter&state=state') + assert response.status_code == 401 + assert response.json['detail'] == "You denied the request to sign in" + + def test_wrong_state_signature_raises(self): + self.auth_type(AUTH_OAUTH) + mock_twitter_auth_provider = mock.Mock() + self.app.appbuilder.sm.oauth_remotes = { + "twitter": mock_twitter_auth_provider, + } + mock_twitter_auth_provider.authorize_access_token.return_value = mock.MagicMock() + response = self.client.get('api/v1/oauth-authorized?provider=twitter&state=state') + assert response.status_code == 400 + assert response.json['detail'] == "State signature is not valid!" + + @mock.patch("airflow.api_connexion.endpoints.auth_endpoint.jwt.decode") + def test_successful_authorization(self, mock_jwt_decode): + self.auth_type(AUTH_OAUTH) + mock_jwt_decode.return_value = {'some': 'payload'} + mock_twitter_auth_provider = mock.Mock() + mock_oauth_session = mock.MagicMock() + mock_user_info = mock.MagicMock() + mock_user_oauth = mock.MagicMock() + self.app.appbuilder.sm.set_oauth_session = mock_oauth_session + self.app.appbuilder.sm.oauth_user_info = mock_user_info + self.app.appbuilder.sm.auth_user_oauth = mock_user_oauth + user = self.app.appbuilder.sm.find_user(username='test') + self.app.appbuilder.sm.auth_user_oauth.return_value = user + self.app.appbuilder.sm.oauth_remotes = { + "twitter": mock_twitter_auth_provider, + } + mock_authorized = mock.MagicMock() + mock_twitter_auth_provider.authorize_access_token.return_value = mock_authorized + self.client.get('api/v1/oauth-authorized?provider=twitter&state=state') + mock_oauth_session.assert_not_called() + mock_user_info.assert_called_once_with('twitter', mock_authorized) + mock_user_oauth.assert_called_once_with(mock_user_info.return_value) + + +class TestRemoteUserLoginEndpoint(TestLoginEndpoint): + def test_remote_user_can_login(self): + self.auth_type(AUTH_REMOTE_USER) + response = self.client.get('api/v1/auth-remoteuser', environ_overrides={"REMOTE_USER": "test"}) + assert response.status_code == 200 + assert response.json['user']['username'] == 'test' + + def test_incorrect_username_raises(self): + self.auth_type(AUTH_REMOTE_USER) + self.app.config['AUTH_USER_REGISTRATION'] = False + response = self.client.get('api/v1/auth-remoteuser', environ_overrides={"REMOTE_USER": "fakeuser"}) + assert response.status_code == 401 + assert response.json['detail'] == 'Invalid login' + + def test_incorrect_auth_type_raises(self): + self.auth_type(AUTH_DB) + resp = self.client.get('api/v1/auth-remoteuser', environ_overrides={"REMOTE_USER": "tes"}) + assert resp.status_code == 401 + assert resp.json['detail'] == AUTH_TYPE_MISMATCH_MESSAGE + + +class TestRefreshTokenEndpoint(TestLoginEndpoint): + def test_creates_access_token(self): + self.auth_type(AUTH_DB) + payload = {"username": "test", "password": "test"} + response = self.client.post('api/v1/auth/login', json=payload) + refresh = response.json['refresh_token'] + response2 = self.client.get("api/v1/refresh", headers={"Authorization": f"Bearer {refresh}"}) + + assert response2.json['access_token'] is not None + + def test_access_token_cant_access_endpoint(self): + self.auth_type(AUTH_DB) + payload = {"username": "test", "password": "test"} + response = self.client.post('api/v1/auth/login', json=payload) + token = response.json['token'] + response2 = self.client.get("api/v1/refresh", headers={"Authorization": f"Bearer {token}"}) + assert response2.status_code == 422 + assert response2.json['msg'] == 'Only refresh tokens are allowed' + + +class TestRevokeAccessTokenEndpoint(TestLoginEndpoint): + @provide_session + def test_revoke_current_user_access_token_works(self, session): + self.auth_type(AUTH_DB) + payload = {"username": "test", "password": "test"} + response = self.client.post('api/v1/auth/login', json=payload) + token = response.json['token'] + response2 = self.client.get( + "api/v1/revoke_access_token", headers={"Authorization": f"Bearer {token}"} + ) + assert response2.json == {'revoked': True} + total_tokens_in_db = session.query(func.count(TokenBlockList.jti)).scalar() + assert total_tokens_in_db == 1 + + +class TestRevokeRefreshTokenEndpoint(TestLoginEndpoint): + @provide_session + def test_revoke_current_user_access_token_works(self, session): + self.auth_type(AUTH_DB) + payload = {"username": "test", "password": "test"} + response = self.client.post('api/v1/auth/login', json=payload) + refresh_token = response.json['refresh_token'] + response2 = self.client.get( + "api/v1/revoke_refresh_token", headers={"Authorization": f"Bearer {refresh_token}"} + ) + assert response2.json == {'revoked': True} + total_tokens_in_db = session.query(func.count(TokenBlockList.jti)).scalar() + assert total_tokens_in_db == 1 + + +class TestTokenRevokeEndpoint(TestLoginEndpoint): + @provide_session + def test_revoke_token_works(self, session): + self.auth_type(AUTH_DB) + payload = {"username": "test", "password": "test"} + response = self.client.post('api/v1/auth/login', json=payload) + token = response.json['token'] + refresh = response.json['refresh_token'] + self.client.post( + "api/v1/revoke", json={"token": refresh}, headers={"Authorization": f"Bearer {token}"} + ) + total_tokens_in_db = session.query(func.count(TokenBlockList.jti)).scalar() + assert total_tokens_in_db == 1 diff --git a/tests/models/test_auth.py b/tests/models/test_auth.py new file mode 100644 index 0000000000000..8ddeed0e6e177 --- /dev/null +++ b/tests/models/test_auth.py @@ -0,0 +1,63 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +import unittest +from datetime import datetime + +from airflow.models.auth import TokenBlockList +from airflow.utils.session import provide_session + +EXPIRY_DELTA = datetime.timestamp(datetime.now()) + + +class TestToken(unittest.TestCase): + def tearDown(self) -> None: + self.delete_tokens() + + @provide_session + def delete_tokens(self, session=None): + tokens = session.query(TokenBlockList).all() + for i in tokens: + session.delete(i) + + @provide_session + def test_add_token(self, session=None): + TokenBlockList.add_token(jti="token", expiry_delta=EXPIRY_DELTA) + token = session.query(TokenBlockList).all() + assert len(token) == 1 + + def test_get_token_method(self): + TokenBlockList.add_token(jti="token", expiry_delta=EXPIRY_DELTA) + token2 = TokenBlockList.get_token("token") + assert token2.jti == "token" + + def test_delete_token_method(self): + TokenBlockList.add_token(jti="token", expiry_delta=EXPIRY_DELTA) + token = TokenBlockList.get_token("token") + assert token is not None + TokenBlockList.delete_token("token") + token = TokenBlockList.get_token("token") + assert token is None + + def test_delete_expired_tokens(self): + TokenBlockList.add_token(jti="token", expiry_delta=EXPIRY_DELTA) + TokenBlockList.add_token(jti="token2", expiry_delta=EXPIRY_DELTA) + assert TokenBlockList.get_token('token') + assert TokenBlockList.get_token('token2') + TokenBlockList.delete_expired_tokens() + assert TokenBlockList.get_token('token') is None + assert TokenBlockList.get_token('token2') is None