[AIRFLOW-489] Add API Framework

This implements a framework for API calls to Airflow. Currently
all access is done by cli or web ui. Especially in the context
of the cli this raises security concerns which can be alleviated
with a secured API call over the wire.

Secondly integration with other systems is a bit harder if you have
to call a cli. For public facing endpoints JSON is used.

As an example the trigger_dag functionality is now made into a
API call.

Backwards compat is retained by switching to a LocalClient.

.rat-excludes

@@ -19,3 +19,8 @@ metastore_db
 .*csv
 CHANGELOG.txt
 .*zip
+# Apache Rat does not detect BSD-2 clause properly
+# it is compatible according to http://www.apache.org/legal/resolved.html#category-a
+kerberos_auth.py
+airflow_api_auth_backend_kerberos_auth_py.html
+

.travis.yml

@@ -28,13 +28,18 @@ addons:
       - mysql-client-core-5.6
       - mysql-client-5.6
       - krb5-user
+      - krb5-kdc
+      - krb5-admin-server
+      - python-selinux
   postgresql: "9.2"
 python:
   - "2.7"
   - "3.4"
 env:
   global:
     - TRAVIS_CACHE=$HOME/.travis_cache/
+    - KRB5_CONFIG=/etc/krb5.conf
+    - KRB5_KTNAME=/etc/airflow.keytab
     # Travis on google cloud engine has a global /etc/boto.cfg that
     # does not work with python 3
     - BOTO_CONFIG=/tmp/bogusvalue

NOTICE

@@ -15,3 +15,4 @@ This product includes Underscore.js (http://underscorejs.org - MIT license), Cop
 This product includes FooTable (http://fooplugins.com/plugins/footable-jquery/ - MIT license), Copyright 2013 Steven Usher & Brad Vincent.
 This product includes dagre (https://github.com/cpettitt/dagre - MIT license), Copyright (c) 2012-2014 Chris Pettitt.
 This product includes d3js (https://d3js.org/ - https://github.com/mbostock/d3/blob/master/LICENSE), Copyright (c) 2010-2016, Michael Bostock.
+This product includes flask-kerberos (https://github.com/mkomitee/flask-kerberos - BSD License), Copyright (c) 2013, Michael Komitee

airflow/api/__init__.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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 __future__ import print_function
+
+import logging
+
+from airflow.exceptions import AirflowException
+from airflow import configuration as conf
+from importlib import import_module
+
+api_auth = None
+
+
+def load_auth():
+    auth_backend = 'airflow.api.auth.backend.default'
+    try:
+        auth_backend = conf.get("api", "auth_backend")
+    except conf.AirflowConfigException:
+        pass
+
+    try:
+        global api_auth
+        api_auth = import_module(auth_backend)
+    except ImportError as err:
+        logging.critical("Cannot import {} for API authentication due to: {}"
+                         .format(auth_backend, err))
+        raise AirflowException(err)

airflow/api/auth/__init__.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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.

airflow/api/auth/backend/__init__.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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.

airflow/api/auth/backend/default.py

@@ -0,0 +1,29 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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 functools import wraps
+
+client_auth = None
+
+
+def init_app(app):
+    pass
+
+
+def requires_authentication(function):
+    @wraps(function)
+    def decorated(*args, **kwargs):
+        return function(*args, **kwargs)
+
+    return decorated

airflow/api/auth/backend/kerberos_auth.py

@@ -0,0 +1,128 @@
+# Copyright (c) 2013, Michael Komitee
+# All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without modification,
+# are permitted provided that the following conditions are met:
+#
+# 1. Redistributions of source code must retain the above copyright notice,
+# this list of conditions and the following disclaimer.
+#
+# 2. Redistributions in binary form must reproduce the above copyright notice,
+# this list of conditions and the following disclaimer in the documentation
+# and/or other materials provided with the distribution.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
+# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+from future.standard_library import install_aliases
+install_aliases()
+
+import kerberos
+import logging
+import os
+
+from airflow import configuration as conf
+
+from flask import Response
+from flask import _request_ctx_stack as stack
+from flask import make_response
+from flask import request
+from flask import g
+from functools import wraps
+
+from requests_kerberos import HTTPKerberosAuth
+from socket import getfqdn
+
+client_auth = HTTPKerberosAuth(service='airflow')
+
+_SERVICE_NAME = None
+
+
+def init_app(app):
+    global _SERVICE_NAME
+
+    hostname = app.config.get('SERVER_NAME')
+    if not hostname:
+        hostname = getfqdn()
+    logging.info("Kerberos: hostname {}".format(hostname))
+
+    service = 'airflow'
+
+    _SERVICE_NAME = "{}@{}".format(service, hostname)
+
+    if 'KRB5_KTNAME' not in os.environ:
+        os.environ['KRB5_KTNAME'] = conf.get('kerberos', 'keytab')
+
+    try:
+        logging.info("Kerberos init: {} {}".format(service, hostname))
+        principal = kerberos.getServerPrincipalDetails(service, hostname)
+    except kerberos.KrbError as err:
+        logging.warn("Kerberos: {}".format(err))
+    else:
+        logging.info("Kerberos API: server is {}".format(principal))
+
+
+def _unauthorized():
+    """
+    Indicate that authorization is required
+    :return:
+    """
+    return Response("Unauthorized", 401, {"WWW-Authenticate": "Negotiate"})
+
+
+def _forbidden():
+    return Response("Forbidden", 403)
+
+
+def _gssapi_authenticate(token):
+    state = None
+    ctx = stack.top
+    try:
+        rc, state = kerberos.authGSSServerInit(_SERVICE_NAME)
+        if rc != kerberos.AUTH_GSS_COMPLETE:
+            return None
+        rc = kerberos.authGSSServerStep(state, token)
+        if rc == kerberos.AUTH_GSS_COMPLETE:
+            ctx.kerberos_token = kerberos.authGSSServerResponse(state)
+            ctx.kerberos_user = kerberos.authGSSServerUserName(state)
+            return rc
+        elif rc == kerberos.AUTH_GSS_CONTINUE:
+            return kerberos.AUTH_GSS_CONTINUE
+        else:
+            return None
+    except kerberos.GSSError:
+        return None
+    finally:
+        if state:
+            kerberos.authGSSServerClean(state)
+
+
+def requires_authentication(function):
+    @wraps(function)
+    def decorated(*args, **kwargs):
+        header = request.headers.get("Authorization")
+        if header:
+            ctx = stack.top
+            token = ''.join(header.split()[1:])
+            rc = _gssapi_authenticate(token)
+            if rc == kerberos.AUTH_GSS_COMPLETE:
+                g.user = ctx.kerberos_user
+                response = function(*args, **kwargs)
+                response = make_response(response)
+                if ctx.kerberos_token is not None:
+                    response.headers['WWW-Authenticate'] = ' '.join(['negotiate',
+                                                                     ctx.kerberos_token])
+
+                return response
+            elif rc != kerberos.AUTH_GSS_CONTINUE:
+                return _forbidden()
+        return _unauthorized()
+    return decorated

airflow/api/client/__init__.py

@@ -0,0 +1,14 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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.
+

airflow/api/client/api_client.py

@@ -0,0 +1,30 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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.
+#
+
+
+class Client:
+    def __init__(self, api_base_url, auth):
+        self._api_base_url = api_base_url
+        self._auth = auth
+
+    def trigger_dag(self, dag_id, run_id=None, conf=None):
+        """
+        Creates a dag run for the specified dag
+        :param dag_id:
+        :param run_id:
+        :param conf:
+        :return:
+        """
+        raise NotImplementedError()

airflow/api/client/json_client.py

@@ -0,0 +1,39 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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 future.moves.urllib.parse import urljoin
+
+from airflow.api.client import api_client
+
+import requests
+
+
+class Client(api_client.Client):
+    def trigger_dag(self, dag_id, run_id=None, conf=None):
+        endpoint = '/api/experimental/dags/{}/dag_runs'.format(dag_id)
+        url = urljoin(self._api_base_url, endpoint)
+
+        resp = requests.post(url,
+                             auth=self._auth,
+                             json={
+                                 "run_id": run_id,
+                                 "conf": conf,
+                             })
+
+        if not resp.ok:
+            raise IOError()
+
+        data = resp.json()
+
+        return data['message']

airflow/api/client/local_client.py

@@ -0,0 +1,22 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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 airflow.api.client import api_client
+from airflow.api.common.experimental import trigger_dag
+
+
+class Client(api_client.Client):
+    def trigger_dag(self, dag_id, run_id=None, conf=None):
+        dr = trigger_dag.trigger_dag(dag_id=dag_id, run_id=run_id, conf=conf)
+        return "Created {}".format(dr)

airflow/api/common/__init__.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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.

airflow/api/common/experimental/__init__.py

@@ -0,0 +1,13 @@
+# -*- coding: utf-8 -*-
+#
+# Licensed 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.