Repository: jpadilla/pyjwt

Issue:
Malformed JWTs due to different signing algorithms in `headers` and `algorithm` parameters
Generating JWTs (`encode()`) with `alg` key set in `headers` parameter along with a different `algorithm` parameter value results in malformed JWTs. 

Example `headers` and `algorithm` values that will result in malformed tokens: 
```python
headers = {"typ": "JWT", "alg": "none"}
algorithm = "HS256"
```

## Expected Result
A token signed with let's say HS256 algorithm should always contain `alg` value set as HS256 in the JWT header. 

## Actual Result
Tokens can be signed using an algorithm x (eg. HS256) with token header containing a different `alg` value

## Reproduction Steps

```python
import jwt
payload = {"test": "payload"}
headers = {"typ": "JWT", "alg": "none"}
print(jwt.encode(payload = payload, key = "secret", algorithm = "HS256", headers = headers))
```
Running the snippet above will give: `eyJ0eXAiOiJKV1QiLCJhbGciOiJub25lIn0.eyJ0ZXN0IjoicGF5bG9hZCJ9.PySJLJ2Js8Z1K0acpZrgOhzHp0sea_N5rrNX1L_FJis`. Header + payload in this case is signed using HS256, but the header value in the token is: `{"typ":"JWT","alg":"none"}`

Similarly,
```python
import jwt
payload = {"test": "payload"}
headers = {"typ": "JWT", "alg": "RS256"}
jwt.encode(payload = payload, key = "secret", algorithm = "HS256", headers = headers)
```
will result in a token signed using HS256, with the token header being `{"typ":"JWT","alg":"RS256"}`

## System Information

    $ python -m jwt.help

```
{
  "cryptography": {
    "version": "3.4.7"
  },
  "implementation": {
    "name": "CPython",
    "version": "3.8.8"
  },
  "platform": {
    "release": "5.8.0-53-generic",
    "system": "Linux"
  },
  "pyjwt": {
    "version": "2.1.0"
  }
}
```

## Root cause 
`algorithm` ([L#109 api_jws.py](https://github.com/jpadilla/pyjwt/blob/master/jwt/api_jws.py#L109)) variable is being used for signing, but `header` dict ([L#93 api_jws.py](https://github.com/jpadilla/pyjwt/blob/master/jwt/api_jws.py#L93)) is being overwritten with `headers` method parameter in [L#97 api_jws.py](https://github.com/jpadilla/pyjwt/blob/master/jwt/api_jws.py#L97) leading to `alg` value being overwritten to the value present in the method parameter `headers`. 

I tried coming up with ways in which this bug could be abused by an adversary, but couldn't find any. And so, I don't think this qualifies as a security issue. 

Discussion:

Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 632d7f9..505e374 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -18,7 +18,7 @@ repos:
         language_version: python3.8
 
   - repo: https://github.com/PyCQA/isort
-    rev: 5.9.2
+    rev: 5.9.3
     hooks:
       - id: isort
 
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 763822a..b049e9e 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -13,6 +13,7 @@ Changed
 Fixed
 ~~~~~
 
+- Prefer `headers["alg"]` to `algorithm` in `jwt.encode()`. `#673 <https://github.com/jpadilla/pyjwt/pull/673>`__
 - Fix aud validation to support {'aud': null} case. `#670 <https://github.com/jpadilla/pyjwt/pull/670>`__
 
 Added
diff --git a/docs/api.rst b/docs/api.rst
index d23d4bd..f1720b9 100644
--- a/docs/api.rst
+++ b/docs/api.rst
@@ -13,8 +13,9 @@ API Reference
         * for **asymmetric algorithms**: PEM-formatted private key, a multiline string
         * for **symmetric algorithms**: plain string, sufficiently long for security
 
-    :param str algorithm: algorithm to sign the token with, e.g. ``"ES256"``
-    :param dict headers: additional JWT header fields, e.g. ``dict(kid="my-key-id")``
+    :param str algorithm: algorithm to sign the token with, e.g. ``"ES256"``.
+        If ``headers`` includes ``alg``, it will be preferred to this parameter.
+    :param dict headers: additional JWT header fields, e.g. ``dict(kid="my-key-id")``.
     :param json.JSONEncoder json_encoder: custom JSON encoder for ``payload`` and ``headers``
     :rtype: str
     :returns: a JSON Web Token
diff --git a/jwt/api_jws.py b/jwt/api_jws.py
index 6d13699..6cefb2c 100644
--- a/jwt/api_jws.py
+++ b/jwt/api_jws.py
@@ -77,7 +77,7 @@ class PyJWS:
         self,
         payload: bytes,
         key: str,
-        algorithm: str = "HS256",
+        algorithm: Optional[str] = "HS256",
         headers: Optional[Dict] = None,
         json_encoder: Optional[Type[json.JSONEncoder]] = None,
     ) -> str:
@@ -86,6 +86,10 @@ class PyJWS:
         if algorithm is None:
             algorithm = "none"
 
+        # Prefer headers["alg"] if present to algorithm parameter.
+        if headers and "alg" in headers and headers["alg"]:
+            algorithm = headers["alg"]
+
         if algorithm not in self._valid_algs:
             pass
 
diff --git a/jwt/api_jwt.py b/jwt/api_jwt.py
index 55a8f29..48a9316 100644
--- a/jwt/api_jwt.py
+++ b/jwt/api_jwt.py
@@ -38,7 +38,7 @@ class PyJWT:
         self,
         payload: Dict[str, Any],
         key: str,
-        algorithm: str = "HS256",
+        algorithm: Optional[str] = "HS256",
         headers: Optional[Dict] = None,
         json_encoder: Optional[Type[json.JSONEncoder]] = None,
     ) -> str:


Test patch:
diff --git a/tests/test_api_jws.py b/tests/test_api_jws.py
index b731edc..fbeda02 100644
--- a/tests/test_api_jws.py
+++ b/tests/test_api_jws.py
@@ -166,6 +166,32 @@ class TestJWS:
         exception = context.value
         assert str(exception) == "Algorithm not supported"
 
+    def test_encode_with_headers_alg_none(self, jws, payload):
+        msg = jws.encode(payload, key=None, headers={"alg": "none"})
+        with pytest.raises(DecodeError) as context:
+            jws.decode(msg, algorithms=["none"])
+        assert str(context.value) == "Signature verification failed"
+
+    @crypto_required
+    def test_encode_with_headers_alg_es256(self, jws, payload):
+        with open(key_path("testkey_ec.priv"), "rb") as ec_priv_file:
+            priv_key = load_pem_private_key(ec_priv_file.read(), password=None)
+        with open(key_path("testkey_ec.pub"), "rb") as ec_pub_file:
+            pub_key = load_pem_public_key(ec_pub_file.read())
+
+        msg = jws.encode(payload, priv_key, headers={"alg": "ES256"})
+        assert b"hello world" == jws.decode(msg, pub_key, algorithms=["ES256"])
+
+    @crypto_required
+    def test_encode_with_alg_hs256_and_headers_alg_es256(self, jws, payload):
+        with open(key_path("testkey_ec.priv"), "rb") as ec_priv_file:
+            priv_key = load_pem_private_key(ec_priv_file.read(), password=None)
+        with open(key_path("testkey_ec.pub"), "rb") as ec_pub_file:
+            pub_key = load_pem_public_key(ec_pub_file.read())
+
+        msg = jws.encode(payload, priv_key, algorithm="HS256", headers={"alg": "ES256"})
+        assert b"hello world" == jws.decode(msg, pub_key, algorithms=["ES256"])
+
     def test_decode_algorithm_param_should_be_case_sensitive(self, jws):
         example_jws = (
             "eyJhbGciOiJoczI1NiIsInR5cCI6IkpXVCJ9"  # alg = hs256