Q:

Airflow SSH operator how to specify authentication type

I am trying to ssh in to one of my servers from Airflow worker using SSHOperator. My SSH is being configured to use authentication type as Kerberos. I get following error with default SSH connection configured with below extra argument.

SSH operator error: Bad authentication type; allowed types: ['publickey,'gssapi-keyex',gssapi-with-mic,'keyboard-interactive']

How can I solve this error. I tried below settings in extras field of my connection in airflow UI.
{ "gss-auth":"true","gssapi-keyex":"true","gss-kex":"true"}

Is there an option in Airflow SSHoperator to specify authentication type to be used is Kerberos? 

A:

There was no Kerberos authentication support in existing SSHOperator of Airflow even if the underlying Paramiko library has that support.
I was able to fix this by writing a custom hook extending SSHHook which passes an argument to the underlying Paramiko library to specify Kerberos as authentication type. It worked! Thanks to Airflow's ease of extensibility.
Custom hook: 
class CustomSSHHook(SSHHook):
    """
    Custom SSH Hook with kerberose authentication support
    """

    def __init__(self,
                 ssh_conn_id=None,
                 remote_host=None,
                 username=None,
                 password=None,
                 key_file=None,
                 port=None,
                 timeout=10,
                 keepalive_interval=30):
        super(CustomSSHHook, self).__init__(
                 ssh_conn_id,
                 remote_host,
                 username,
                 password,
                 key_file,
                 port,
                 timeout,
                 keepalive_interval)
        # Use connection to override defaults
        self.gss_auth = False
        if self.ssh_conn_id is not None:
            conn = self.get_connection(self.ssh_conn_id)

            if conn.extra is not None:
                extra_options = conn.extra_dejson
                if "gss_auth" in extra_options \
                        and str(extra_options["gss_auth"]).lower() == 'true':
                    self.gss_auth = True

    def get_conn(self):
        """
                Opens a ssh connection to the remote host.

                :return paramiko.SSHClient object
                """

        self.log.debug('Creating SSH client for conn_id: %s', self.ssh_conn_id)
        client = paramiko.SSHClient()
        client.load_system_host_keys()
        if self.no_host_key_check:
            # Default is RejectPolicy
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        if self.password and self.password.strip():
            client.connect(hostname=self.remote_host,
                           username=self.username,
                           password=self.password,
                           key_filename=self.key_file,
                           timeout=self.timeout,
                           compress=self.compress,
                           port=self.port,
                           sock=self.host_proxy)
        else:
            client.connect(hostname=self.remote_host,
                           username=self.username,
                           key_filename=self.key_file,
                           timeout=self.timeout,
                           compress=self.compress,
                           port=self.port,
                           sock=self.host_proxy,
                           gss_auth=self.gss_auth)

        if self.keepalive_interval:
            client.get_transport().set_keepalive(self.keepalive_interval)

        self.client = client
        return client

class CustomSshPlugin(AirflowPlugin):
    name = "ssh_plugins"
    DEFAULT_PLUGIN = "1.0"
    hooks = [CustomSSHHook]

And above can be used in DAG as follows:
from airflow.hooks.ssh_plugins import CustomSSHHook

edge_node_hook = CustomSSHHook(ssh_conn_id="ssh_con_id",
                         remote_host=host_ip,
                         port=22,
                         timeout=100)
ssh_execution = SSHOperator(dag=dag, task_id='sample_task',
                                          ssh_hook=edge_node_hook,
                                          command='whoami ',
                                          do_xcom_push=False)

And you have to add below argument in extras field of connection with id 'ssh_conn_id'.
{"gss_auth":"true"}

