<p dir="auto"><strong>Description</strong><br>
we need a feature where we can set the state of the task instance to the specified state.</p>
<p dir="auto"><strong>Use case / motivation</strong><br>
In my application, some task instances have been scheduled for a long time, we want to put it in the failed state. Currently, we are doing this by using airflow Web UI but we want to use API instead.</p>
<p dir="auto"><strong>Are you willing to submit a PR?</strong><br>
Yes, I'm willing to.</p>
<p dir="auto"><strong>Related Issues</strong><br>
<a class="issue-link js-issue-link" data-error-text="Failed to load title" data-id="919443927" data-permission-text="Title is private" data-url="https://github.com/apache/airflow/issues/16412" data-hovercard-type="issue" data-hovercard-url="/apache/airflow/issues/16412/hovercard" href="https://github.com/apache/airflow/issues/16412">#16412</a></p>
<p dir="auto"><strong>Apache Airflow version</strong>: 2.0.2</p>
<p dir="auto"><strong>Kubernetes version (if you are using kubernetes)</strong> (use <code class="notranslate">kubectl version</code>):</p>
<p dir="auto"><strong>Environment</strong>:</p>
<ul dir="auto">
<li><strong>Cloud provider or hardware configuration</strong>:</li>
<li><strong>OS</strong> (e.g. from /etc/os-release):</li>
<li><strong>Kernel</strong> (e.g. <code class="notranslate">uname -a</code>):</li>
<li><strong>Install tools</strong>:</li>
<li><strong>Others</strong>:</li>
</ul>
<p dir="auto"><strong>What happened</strong>:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="ssh-keygen -t rsa -P &quot;&quot; -f test_rsa
cat test_rsa | python -c 'import sys;import json; print(json.dumps(sys.stdin.read()))' # gives the private key encoded as JSON string to be pasted in the connection extra private_key"><pre class="notranslate"><code class="notranslate">ssh-keygen -t rsa -P "" -f test_rsa
cat test_rsa | python -c 'import sys;import json; print(json.dumps(sys.stdin.read()))' # gives the private key encoded as JSON string to be pasted in the connection extra private_key
</code></pre></div>
<p dir="auto">I created an Airflow Connection</p>
<ul dir="auto">
<li>type: ssh</li>
<li>extra:</li>
</ul>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="{
&quot;look_for_keys&quot;: &quot;false&quot;,
&quot;no_host_key_check&quot;: &quot;true&quot;,
&quot;private_key&quot;: &quot;-----BEGIN OPENSSH PRIVATE KEY-----\nb3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAABlwAAAAdzc2gtcn\nNhAA........W4tTGFndW5hLU1hY0Jvb2stUHJvAQI=\n-----END OPENSSH PRIVATE KEY-----\n&quot;,
&quot;private_key_passphrase&quot;: &quot;&quot;
}"><pre class="notranslate"><code class="notranslate">{
"look_for_keys": "false",
"no_host_key_check": "true",
"private_key": "-----BEGIN OPENSSH PRIVATE KEY-----\nb3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAABlwAAAAdzc2gtcn\nNhAA........W4tTGFndW5hLU1hY0Jvb2stUHJvAQI=\n-----END OPENSSH PRIVATE KEY-----\n",
"private_key_passphrase": ""
}
</code></pre></div>
<p dir="auto">When this SSH connection is used in SFTPToS3Operator for example it will incorrectly parse that <code class="notranslate">private_key</code> as a <code class="notranslate">paramiko.dsskey.DSSKey</code> instead of the correct <code class="notranslate">paramiko.rsakey.RSAKey</code>.</p>
<p dir="auto">The code responsible for the processing of <code class="notranslate">private_key</code> is not <strong>not deterministic</strong> (I don't think <code class="notranslate">.values()</code> returns items in any particular order) , but in my case it will always try <code class="notranslate">paramiko.dsskey.DSSKey</code>  before it tries <code class="notranslate">paramiko.rsakey.RSAKey</code>:</p>
<p dir="auto"></p><div class="Box Box--condensed my-2">
  <div class="Box-header f6">
    <p class="mb-0 text-bold">
      <a href="https://github.com/apache/airflow/blob/8e2a0bc2e39aeaf15b409bbfa8ac0c85aa873815/airflow/providers/ssh/hooks/ssh.py#L363-L369">airflow/airflow/providers/ssh/hooks/ssh.py</a>
    </p>
    <p class="mb-0 color-fg-muted">
        Lines 363 to 369
      in
      <a data-pjax="true" class="commit-tease-sha" href="/apache/airflow/commit/8e2a0bc2e39aeaf15b409bbfa8ac0c85aa873815">8e2a0bc</a>
    </p>
  </div>
  <div itemprop="text" class="Box-body p-0 blob-wrapper blob-wrapper-embedded data">
    <table class="highlight tab-size mb-0 js-file-line-container" data-tab-size="8" data-paste-markdown-skip="">

        <tbody><tr class="border-0">
          <td id="L363" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="363"></td>
          <td id="LC363" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-s1">allowed_pkey_types</span> <span class="pl-c1">=</span> <span class="pl-s1">self</span>.<span class="pl-s1">_default_pkey_mappings</span>.<span class="pl-en">values</span>() </td>
        </tr>

        <tr class="border-0">
          <td id="L364" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="364"></td>
          <td id="LC364" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line"> <span class="pl-k">for</span> <span class="pl-s1">pkey_type</span> <span class="pl-c1">in</span> <span class="pl-s1">allowed_pkey_types</span>: </td>
        </tr>

        <tr class="border-0">
          <td id="L365" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="365"></td>
          <td id="LC365" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-k">try</span>: </td>
        </tr>

        <tr class="border-0">
          <td id="L366" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="366"></td>
          <td id="LC366" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">         <span class="pl-s1">key</span> <span class="pl-c1">=</span> <span class="pl-s1">pkey_type</span>.<span class="pl-en">from_private_key</span>(<span class="pl-v">StringIO</span>(<span class="pl-s1">private_key</span>), <span class="pl-s1">password</span><span class="pl-c1">=</span><span class="pl-s1">passphrase</span>) </td>
        </tr>

        <tr class="border-0">
          <td id="L367" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="367"></td>
          <td id="LC367" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">         <span class="pl-k">return</span> <span class="pl-s1">key</span> </td>
        </tr>

        <tr class="border-0">
          <td id="L368" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="368"></td>
          <td id="LC368" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">     <span class="pl-k">except</span> <span class="pl-s1">paramiko</span>.<span class="pl-s1">ssh_exception</span>.<span class="pl-v">SSHException</span>: </td>
        </tr>

        <tr class="border-0">
          <td id="L369" class="blob-num border-0 px-3 py-0 color-bg-default" data-line-number="369"></td>
          <td id="LC369" class="blob-code border-0 px-3 py-0 color-bg-default blob-code-inner js-file-line">         <span class="pl-k">continue</span> </td>
        </tr>
    </tbody></table>
  </div>
</div>
<p></p>
<p dir="auto">This incorrectly parsed private key will cause a very confusing error  later when it's actually used</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2021-06-30 23:33:14,604] {transport.py:1819} INFO - Connected (version 2.0, client AWS_SFTP_1.0)
[2021-06-30 23:33:14,732] {transport.py:1819} ERROR - Unknown exception: q must be exactly 160, 224, or 256 bits long
[2021-06-30 23:33:14,737] {transport.py:1817} ERROR - Traceback (most recent call last):
[2021-06-30 23:33:14,737] {transport.py:1817} ERROR -   File &quot;/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/paramiko/transport.py&quot;, line 2109, in run
[2021-06-30 23:33:14,737] {transport.py:1817} ERROR -     handler(self.auth_handler, m)
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -   File &quot;/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/paramiko/auth_handler.py&quot;, line 298, in _parse_service_accept
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -     sig = self.private_key.sign_ssh_data(blob)
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -   File &quot;/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/paramiko/dsskey.py&quot;, line 108, in sign_ssh_data
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -     key = dsa.DSAPrivateNumbers(
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -   File &quot;/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/primitives/asymmetric/dsa.py&quot;, line 244, in private_key
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -     return backend.load_dsa_private_numbers(self)
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -   File &quot;/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/backends/openssl/backend.py&quot;, line 826, in load_dsa_private_numbers
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -     dsa._check_dsa_private_numbers(numbers)
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR -   File &quot;/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/primitives/asymmetric/dsa.py&quot;, line 282, in _check_dsa_private_numbers
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR -     _check_dsa_parameters(parameters)
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR -   File &quot;/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/primitives/asymmetric/dsa.py&quot;, line 274, in _check_dsa_parameters
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR -     raise ValueError(&quot;q must be exactly 160, 224, or 256 bits long&quot;)
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR - ValueError: q must be exactly 160, 224, or 256 bits long
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR - "><pre class="notranslate"><code class="notranslate">[2021-06-30 23:33:14,604] {transport.py:1819} INFO - Connected (version 2.0, client AWS_SFTP_1.0)
[2021-06-30 23:33:14,732] {transport.py:1819} ERROR - Unknown exception: q must be exactly 160, 224, or 256 bits long
[2021-06-30 23:33:14,737] {transport.py:1817} ERROR - Traceback (most recent call last):
[2021-06-30 23:33:14,737] {transport.py:1817} ERROR -   File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/paramiko/transport.py", line 2109, in run
[2021-06-30 23:33:14,737] {transport.py:1817} ERROR -     handler(self.auth_handler, m)
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -   File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/paramiko/auth_handler.py", line 298, in _parse_service_accept
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -     sig = self.private_key.sign_ssh_data(blob)
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -   File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/paramiko/dsskey.py", line 108, in sign_ssh_data
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -     key = dsa.DSAPrivateNumbers(
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -   File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/primitives/asymmetric/dsa.py", line 244, in private_key
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -     return backend.load_dsa_private_numbers(self)
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -   File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/backends/openssl/backend.py", line 826, in load_dsa_private_numbers
[2021-06-30 23:33:14,738] {transport.py:1817} ERROR -     dsa._check_dsa_private_numbers(numbers)
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR -   File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/primitives/asymmetric/dsa.py", line 282, in _check_dsa_private_numbers
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR -     _check_dsa_parameters(parameters)
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR -   File "/Users/rubelagu/git/apache-airflow-providers-tdh/venv/lib/python3.8/site-packages/cryptography/hazmat/primitives/asymmetric/dsa.py", line 274, in _check_dsa_parameters
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR -     raise ValueError("q must be exactly 160, 224, or 256 bits long")
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR - ValueError: q must be exactly 160, 224, or 256 bits long
[2021-06-30 23:33:14,739] {transport.py:1817} ERROR - 
</code></pre></div>

<p dir="auto"><strong>What you expected to happen</strong>:</p>
<p dir="auto">I expected to parse the <code class="notranslate">private_key</code> as a RSAKey.</p>
<p dir="auto">I did my own test and <code class="notranslate">paramiko.dsskey.DSSKey.from_private_key(StringIO(private_key), password=passphrase)</code> will happily parse (incorrectly) a RSA key. The current code assumes that it will raise an exception but it won't.</p>

<p dir="auto"><strong>How to reproduce it</strong>:</p>

<p dir="auto"><strong>Anything else we need to know</strong>:</p>
<p dir="auto">For me it happens always, I don't think the  order of <code class="notranslate">.values()</code> is deterministic, but in my laptop it will always try DSSKey before RSAKey.</p>