-
Notifications
You must be signed in to change notification settings - Fork 30
SimpleSAML integration to support FAU's StudOn #127
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from all commits
4301dda
e9eebe0
2c22794
ac7bf13
77c6e7d
8d37f42
afbfea6
2509711
b10b75d
c21a9f0
8f7748d
e59a9db
31d2c86
40715d6
31758e7
5ba56f7
afe5819
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||
|---|---|---|---|---|
| @@ -0,0 +1,121 @@ | ||||
| from typing import Any, Optional, cast | ||||
|
|
||||
| import aiohttp | ||||
| import yarl | ||||
| from bs4 import BeautifulSoup, Tag | ||||
|
|
||||
| from ...auth import Authenticator, TfaAuthenticator | ||||
| from ...logging import log | ||||
| from ...utils import soupify | ||||
| from ..crawler import CrawlError | ||||
|
|
||||
|
|
||||
| class SimpleSAMLLogin: | ||||
| """ | ||||
| Login via a SimpleSAML system. | ||||
|
|
||||
| It performs a basic authentication by following the login redirect | ||||
| and posting credentials to the indicated form. It also supports TFA similar to Shibboleth. | ||||
| """ | ||||
|
|
||||
| def __init__( | ||||
| self, ilias_url: str, authenticator: Authenticator, tfa_authenticator: Optional[Authenticator] | ||||
| ) -> None: | ||||
| self._ilias_url = ilias_url | ||||
| self._auth = authenticator | ||||
| self._tfa_auth = tfa_authenticator | ||||
|
|
||||
| async def login(self, sess: aiohttp.ClientSession) -> None: | ||||
| """ | ||||
| Perform a SimpleSAML login flow and populate the session cookies. | ||||
| """ | ||||
|
|
||||
| # Start at the local login entrypoint which may redirect to SimpleSAML | ||||
| url = f"{self._ilias_url}/saml.php" | ||||
| async with sess.get(url) as response: | ||||
| saml_url = response.url | ||||
| # If the redirect stayed on the ILIAS host, assume we're already logged in | ||||
| if str(saml_url).startswith(self._ilias_url): | ||||
| log.explain("ILIAS recognized our SAML token and logged us in in the background, returning") | ||||
| return | ||||
| soup: BeautifulSoup = soupify(await response.read()) | ||||
|
|
||||
| # The SimpleSAML login page uses a form POST similar to Shibboleth. | ||||
| # Attempt to login using credentials. | ||||
| while not self._login_successful(soup): | ||||
| form = cast(Tag, soup.find("form", {"method": "post"})) | ||||
| action = cast(str, form["action"]) | ||||
| # dynamically determine full URL from action (FAU uses full URL here, KIT uses relative URL) | ||||
| url = action if action.startswith("https") else str(saml_url.origin()) + action | ||||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This sounds like you want a
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yes would be better.
|
||||
|
|
||||
| username, password = await self._auth.credentials() | ||||
| data = { | ||||
| "username": username, | ||||
| "password": password, | ||||
| } | ||||
| if csrf_token_input := form.find("input", {"name": "csrf_token"}): | ||||
| data["csrf_token"] = csrf_token_input["value"] # type: ignore | ||||
|
|
||||
| soup = await _post(sess, url, data) | ||||
|
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The URL returned here doesn't matter, I assume? It is okay to re-use the saml_url from above later on.
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Sorry, I don't quite get your point.
Collaborator
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I was just wondering whether the post also redirects you and you need to use the new URL for your later POSTs. But if it works for you it probably doesn't and the existing code is fine
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ah, I see. |
||||
|
|
||||
| # Detect attribute release prompt | ||||
| if soup.find(id="attributeRelease"): | ||||
| raise CrawlError( | ||||
| "ILIAS SAML entitlements changed! Please log in once in your browser and review them" | ||||
| ) | ||||
|
|
||||
| if self._tfa_required(soup): | ||||
| soup = await self._authenticate_tfa(sess, soup, saml_url) | ||||
|
|
||||
| if not self._login_successful(soup): | ||||
| self._auth.invalidate_credentials() | ||||
|
|
||||
| # Equivalent: Being redirected via JS automatically | ||||
| # (or clicking "Continue" if you have JS disabled) | ||||
| relay_state = cast(Tag, soup.find("input", {"name": "RelayState"})) | ||||
| saml_response = cast(Tag, soup.find("input", {"name": "SAMLResponse"})) | ||||
| url = cast(str, cast(Tag, soup.find("form", {"method": "post"}))["action"]) | ||||
| data = { # using the info obtained in the while loop above | ||||
| "RelayState": cast(str, relay_state["value"]), | ||||
| "SAMLResponse": cast(str, saml_response["value"]), | ||||
| } | ||||
| await sess.post(cast(str, url), data=data) | ||||
|
|
||||
| async def _authenticate_tfa( | ||||
| self, session: aiohttp.ClientSession, soup: BeautifulSoup, saml_url: yarl.URL | ||||
| ) -> BeautifulSoup: | ||||
| if not self._tfa_auth: | ||||
| self._tfa_auth = TfaAuthenticator("ilias-anon-tfa") | ||||
|
|
||||
| tfa_token = await self._tfa_auth.password() | ||||
|
|
||||
| # Searching the form here so that this fails before asking for | ||||
| # credentials rather than after asking. | ||||
| form = cast(Tag, soup.find("form", {"method": "post"})) | ||||
| action = cast(str, form["action"]) | ||||
| # dynamically determine full URL from action (FAU uses full URL here, KIT uses relative URL) | ||||
| url = action if action.startswith("https") else str(saml_url.origin()) + action | ||||
|
NIKL45 marked this conversation as resolved.
|
||||
|
|
||||
| data = { # for www.sso.uni-erlangen.de/simplesaml/module.php/mfa/otp?... | ||||
| "otp": tfa_token | ||||
| } | ||||
| if csrf_token_input := form.find("input", {"name": "csrf_token"}): | ||||
| data["csrf_token"] = csrf_token_input["value"] # type: ignore | ||||
| return await _post(session, url, data) | ||||
|
|
||||
| @staticmethod | ||||
| def _login_successful(soup: BeautifulSoup) -> bool: | ||||
| relay_state = soup.find("input", {"name": "RelayState"}) | ||||
| saml_response = soup.find("input", {"name": "SAMLResponse"}) | ||||
| return relay_state is not None and saml_response is not None | ||||
|
|
||||
| @staticmethod | ||||
| def _tfa_required(soup: BeautifulSoup) -> bool: | ||||
| # Also treat a body with id="mfa:otp" as TFA required (for FAU) | ||||
| body = soup.find("body") | ||||
| return body is not None and body.get("id") == "mfa:otp" | ||||
|
|
||||
|
|
||||
| async def _post(session: aiohttp.ClientSession, url: str, data: Any) -> BeautifulSoup: | ||||
| async with session.post(url, data=data) as response: | ||||
| return soupify(await response.read()) | ||||
Uh oh!
There was an error while loading. Please reload this page.