Coverage for src/bayernwerk_client/map/auth.py: 34%

1"""Interactive login via a real browser (Playwright). 

2 

3bayernwerk-netz.de sits behind a Cloudflare JS challenge, and the login 

4form itself is a Salesforce Experience Cloud (Aura) app rather than a plain 

5HTML form - neither is realistically replicable with plain HTTP requests. 

6A real browser engine is therefore the only robust way to obtain the initial 

7OIDC/PKCE tokens. Everything after login (all icon-api.eon.com calls) uses 

8plain httpx - see client.py. 

9 

10Note on `refresh_token`: `TokenSet.refresh_token` is captured from 

11sessionStorage alongside the access token, but it is a *Salesforce community* 

12refresh token. POSTing it to the standard 

13`https://account.bayernwerk-netz.de/iconnect/services/oauth2/token` endpoint 

14does return HTTP 200 with a fresh `access_token` - but that token is a native 

15Salesforce session token (opaque, not the `eyJraWQi...` JWT format), not the 

16JWT icon-api.eon.com expects. The JWT is minted by a separate, not-yet 

17identified "Funke IAM" exchange step that only the SPA's own JS performs, so 

18there is currently no known way to refresh *that* token without a browser. 

19Access tokens are valid for ~1h; just re-run `login_interactive` when 

20expired (a few seconds) rather than trying to refresh. 

21""" 

22 

23from __future__ import annotations 

24 

25import json 

26import time 

27from typing import TYPE_CHECKING 

28 

29from bayernwerk_client.exceptions import AuthenticationError 

30from bayernwerk_client.tokens import TokenSet 

31 

32if TYPE_CHECKING: 

33 from playwright.sync_api import Page 

34 

35DEFAULT_PORTAL_URL = "https://www.bayernwerk-netz.de/de/meinauftragsportal.html" 

36 

37 

38def login_interactive( 

39 email: str, 

40 password: str, 

41 *, 

42 portal_url: str = DEFAULT_PORTAL_URL, 

43 headless: bool = False, 

44 timeout: float = 60.0, 

45) -> TokenSet: 

46 """Log in through a real (headed by default) browser and return the resulting tokens. 

47 

48 Raises AuthenticationError on wrong credentials, an unexpected page 

49 (e.g. an MFA prompt this library doesn't handle), or a timeout. 

50 """ 

51 try: 

52 from playwright.sync_api import TimeoutError as PlaywrightTimeoutError 

53 from playwright.sync_api import sync_playwright 

54 except ImportError as exc: 

55 raise AuthenticationError( 

56 "Playwright is required for login. Install with: " 

57 "uv pip install 'bayernwerk-client[login]' && playwright install chromium" 

58 ) from exc 

59 

60 with sync_playwright() as playwright: 

61 browser = playwright.chromium.launch(headless=headless) 

62 try: 

63 page = browser.new_page() 

64 page.goto(portal_url, wait_until="domcontentloaded", timeout=timeout * 1000) 

65 

66 page.get_by_label("E-Mail-Adresse").fill(email) 

67 page.get_by_label("Passwort").fill(password) 

68 page.get_by_role("button", name="Anmelden").click() 

69 

70 return _wait_for_tokens(page, timeout=timeout) 

71 except PlaywrightTimeoutError as exc: 

72 raise AuthenticationError( 

73 "Timed out during login (wrong credentials, an unhandled MFA/consent prompt, " 

74 "or the portal's login page changed)." 

75 ) from exc 

76 finally: 

77 browser.close() 

78 

79 

80def _wait_for_tokens(page: Page, *, timeout: float) -> TokenSet: 

81 deadline = time.monotonic() + timeout 

82 while time.monotonic() < deadline: 

83 for frame in page.frames: 

84 try: 

85 raw = frame.evaluate("() => JSON.stringify(window.sessionStorage)") 

86 except Exception: # noqa: S112, BLE001 - frame may be mid-navigation/detached, just retry 

87 continue 

88 if not raw: 

89 continue 

90 items: dict[str, str] = json.loads(raw) 

91 access_token = _find(items, include="token_store", exclude=("refresh", "idtoken", "verifier")) 

92 if access_token: 

93 refresh_token = _find(items, include="refresh_token_store") 

94 id_token = _find(items, include="idtoken") 

95 return TokenSet.from_access_token(access_token, refresh_token=refresh_token, id_token=id_token) 

96 page.wait_for_timeout(500) 

97 raise AuthenticationError("Login did not produce an access token within the timeout") 

98 

99 

100def _find(items: dict[str, str], *, include: str, exclude: tuple[str, ...] = ()) -> str | None: 

101 for key, value in items.items(): 

102 key_lower = key.lower() 

103 if include not in key_lower: 

104 continue 

105 if any(term in key_lower for term in exclude): 

106 continue 

107 if value: 

108 return value 

109 return None