6
6
import urllib .error
7
7
import urllib .parse
8
8
import urllib .request
9
- from os import environ
10
9
from typing import TYPE_CHECKING
11
10
12
11
from aiohttp import ClientSession
@@ -31,9 +30,12 @@ def get_websocket_url(url: str) -> str:
31
30
def is_supervisor () -> bool :
32
31
"""Return if we're running inside the HA Supervisor (e.g. HAOS)."""
33
32
try :
34
- urllib .request .urlopen ("http://supervisor/core/api " , timeout = 0.5 )
33
+ urllib .request .urlopen ("http://supervisor/core" , timeout = 1 )
35
34
except urllib .error .URLError as err :
36
- return err .reason == "Unauthorized" and environ .get ("HASSIO_TOKEN" ) is not None
35
+ # this should return a 401 unauthorized if it exists
36
+ return getattr (err , "code" , 999 ) == 401
37
+ except TimeoutError :
38
+ return False
37
39
return False
38
40
39
41
@@ -44,7 +46,10 @@ async def async_is_supervisor() -> bool:
44
46
45
47
46
48
def get_auth_url (
47
- hass_url : str , redirect_uri : str , client_id : str | None = None , state : str | None = None
49
+ hass_url : str ,
50
+ redirect_uri : str ,
51
+ client_id : str | None = None ,
52
+ state : str | None = None ,
48
53
) -> str :
49
54
"""
50
55
Return URL to auth flow.
@@ -100,7 +105,9 @@ async def get_token(
100
105
101
106
async with ClientSession () as session :
102
107
resp = await session .post (
103
- url , data = body , headers = {"Content-Type" : "application/x-www-form-urlencoded" }
108
+ url ,
109
+ data = body ,
110
+ headers = {"Content-Type" : "application/x-www-form-urlencoded" },
104
111
)
105
112
result = await resp .json ()
106
113
if "error" in result :
0 commit comments