Long-lived access token (#16453)

* Allow create refresh_token with specific access_token_expiration

* Add token_type, client_name and client_icon

* Add unit test

* Add websocket API to create long-lived access token

* Allow URL use as client_id for long-lived access token

* Remove mutate_refresh_token method

* Use client name as id for long_lived_access_token type refresh token

* Minor change

* Do not allow duplicate client name

* Update docstring

* Remove unnecessary `list`

tests/auth/test_init.py

@@ -2,6 +2,7 @@
 from datetime import timedelta
 from unittest.mock import Mock, patch
 
+import jwt
 import pytest
 import voluptuous as vol
 
@@ -323,7 +324,7 @@ async def test_generating_system_user(hass):
 
 
 async def test_refresh_token_requires_client_for_user(hass):
-    """Test that we can add a system user."""
+    """Test create refresh token for a user with client_id."""
     manager = await auth.auth_manager_from_config(hass, [], [])
     user = MockUser().add_to_auth_manager(manager)
     assert user.system_generated is False
@@ -334,10 +335,14 @@ async def test_refresh_token_requires_client_for_user(hass):
     token = await manager.async_create_refresh_token(user, CLIENT_ID)
     assert token is not None
     assert token.client_id == CLIENT_ID
+    assert token.token_type == auth_models.TOKEN_TYPE_NORMAL
+    # default access token expiration
+    assert token.access_token_expiration == \
+        auth_const.ACCESS_TOKEN_EXPIRATION
 
 
 async def test_refresh_token_not_requires_client_for_system_user(hass):
-    """Test that we can add a system user."""
+    """Test create refresh token for a system user w/o client_id."""
     manager = await auth.auth_manager_from_config(hass, [], [])
     user = await manager.async_create_system_user('Hass.io')
     assert user.system_generated is True
@@ -348,6 +353,56 @@ async def test_refresh_token_not_requires_client_for_system_user(hass):
     token = await manager.async_create_refresh_token(user)
     assert token is not None
     assert token.client_id is None
+    assert token.token_type == auth_models.TOKEN_TYPE_SYSTEM
+
+
+async def test_refresh_token_with_specific_access_token_expiration(hass):
+    """Test create a refresh token with specific access token expiration."""
+    manager = await auth.auth_manager_from_config(hass, [], [])
+    user = MockUser().add_to_auth_manager(manager)
+
+    token = await manager.async_create_refresh_token(
+        user, CLIENT_ID,
+        access_token_expiration=timedelta(days=100))
+    assert token is not None
+    assert token.client_id == CLIENT_ID
+    assert token.access_token_expiration == timedelta(days=100)
+
+
+async def test_refresh_token_type(hass):
+    """Test create a refresh token with token type."""
+    manager = await auth.auth_manager_from_config(hass, [], [])
+    user = MockUser().add_to_auth_manager(manager)
+
+    with pytest.raises(ValueError):
+        await manager.async_create_refresh_token(
+            user, CLIENT_ID, token_type=auth_models.TOKEN_TYPE_SYSTEM)
+
+    token = await manager.async_create_refresh_token(
+        user, CLIENT_ID,
+        token_type=auth_models.TOKEN_TYPE_NORMAL)
+    assert token is not None
+    assert token.client_id == CLIENT_ID
+    assert token.token_type == auth_models.TOKEN_TYPE_NORMAL
+
+
+async def test_refresh_token_type_long_lived_access_token(hass):
+    """Test create a refresh token has long-lived access token type."""
+    manager = await auth.auth_manager_from_config(hass, [], [])
+    user = MockUser().add_to_auth_manager(manager)
+
+    with pytest.raises(ValueError):
+        await manager.async_create_refresh_token(
+            user, token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN)
+
+    token = await manager.async_create_refresh_token(
+        user, client_name='GPS LOGGER', client_icon='mdi:home',
+        token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN)
+    assert token is not None
+    assert token.client_id is None
+    assert token.client_name == 'GPS LOGGER'
+    assert token.client_icon == 'mdi:home'
+    assert token.token_type == auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
 
 
 async def test_cannot_deactive_owner(mock_hass):
@@ -378,6 +433,88 @@ async def test_remove_refresh_token(mock_hass):
     )
 
 
+async def test_create_access_token(mock_hass):
+    """Test normal refresh_token's jwt_key keep same after used."""
+    manager = await auth.auth_manager_from_config(mock_hass, [], [])
+    user = MockUser().add_to_auth_manager(manager)
+    refresh_token = await manager.async_create_refresh_token(user, CLIENT_ID)
+    assert refresh_token.token_type == auth_models.TOKEN_TYPE_NORMAL
+    jwt_key = refresh_token.jwt_key
+    access_token = manager.async_create_access_token(refresh_token)
+    assert access_token is not None
+    assert refresh_token.jwt_key == jwt_key
+    jwt_payload = jwt.decode(access_token, jwt_key, algorithm=['HS256'])
+    assert jwt_payload['iss'] == refresh_token.id
+    assert jwt_payload['exp'] - jwt_payload['iat'] == \
+        timedelta(minutes=30).total_seconds()
+
+
+async def test_create_long_lived_access_token(mock_hass):
+    """Test refresh_token's jwt_key changed for long-lived access token."""
+    manager = await auth.auth_manager_from_config(mock_hass, [], [])
+    user = MockUser().add_to_auth_manager(manager)
+    refresh_token = await manager.async_create_refresh_token(
+        user, client_name='GPS Logger',
+        token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
+        access_token_expiration=timedelta(days=300))
+    assert refresh_token.token_type == \
+        auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
+    access_token = manager.async_create_access_token(refresh_token)
+    jwt_payload = jwt.decode(
+        access_token, refresh_token.jwt_key, algorithm=['HS256'])
+    assert jwt_payload['iss'] == refresh_token.id
+    assert jwt_payload['exp'] - jwt_payload['iat'] == \
+        timedelta(days=300).total_seconds()
+
+
+async def test_one_long_lived_access_token_per_refresh_token(mock_hass):
+    """Test one refresh_token can only have one long-lived access token."""
+    manager = await auth.auth_manager_from_config(mock_hass, [], [])
+    user = MockUser().add_to_auth_manager(manager)
+    refresh_token = await manager.async_create_refresh_token(
+        user, client_name='GPS Logger',
+        token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
+        access_token_expiration=timedelta(days=3000))
+    assert refresh_token.token_type == \
+        auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
+    access_token = manager.async_create_access_token(refresh_token)
+    jwt_key = refresh_token.jwt_key
+
+    rt = await manager.async_validate_access_token(access_token)
+    assert rt.id == refresh_token.id
+
+    with pytest.raises(ValueError):
+        await manager.async_create_refresh_token(
+            user, client_name='GPS Logger',
+            token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
+            access_token_expiration=timedelta(days=3000))
+
+    await manager.async_remove_refresh_token(refresh_token)
+    assert refresh_token.id not in user.refresh_tokens
+    rt = await manager.async_validate_access_token(access_token)
+    assert rt is None, 'Previous issued access token has been invoked'
+
+    refresh_token_2 = await manager.async_create_refresh_token(
+        user, client_name='GPS Logger',
+        token_type=auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN,
+        access_token_expiration=timedelta(days=3000))
+    assert refresh_token_2.id != refresh_token.id
+    assert refresh_token_2.token_type == \
+        auth_models.TOKEN_TYPE_LONG_LIVED_ACCESS_TOKEN
+    access_token_2 = manager.async_create_access_token(refresh_token_2)
+    jwt_key_2 = refresh_token_2.jwt_key
+
+    assert access_token != access_token_2
+    assert jwt_key != jwt_key_2
+
+    rt = await manager.async_validate_access_token(access_token_2)
+    jwt_payload = jwt.decode(
+        access_token_2, rt.jwt_key, algorithm=['HS256'])
+    assert jwt_payload['iss'] == refresh_token_2.id
+    assert jwt_payload['exp'] - jwt_payload['iat'] == \
+        timedelta(days=3000).total_seconds()
+
+
 async def test_login_with_auth_module(mock_hass):
     """Test login as existing user with auth module."""
     manager = await auth.auth_manager_from_config(mock_hass, [{