Merge pull request #10 from numerique-gouv/check-affiliation

Only allow employees to authenticate

CHANGELOG.md

@@ -6,6 +6,9 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0),
 and this project adheres to
 [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+- require "employee" value in eduPersonAffiliation
+
 ## [1.0.1] - 2024-06-04
 - enable JSON logging and LOG_LEVEL environment variable
 - add production configuration

docker/oidc-test-client/app.py

@@ -68,6 +68,12 @@ def oidc_callback():
 
     assert aresp["state"] == session["state"]
 
+    if "error" in aresp:
+        return jsonify(
+            error_response=aresp.to_dict(),
+        )
+
+
     code = aresp["code"]
     logging.info("got auth code=%s", code)
 

src/satosa/internal_attributes.yaml

@@ -25,6 +25,9 @@ attributes:
   eppn:
     saml:
       - eduPersonPrincipalName
+  eduPersonAffiliation:
+    saml:
+      - eduPersonAffiliation
   uid:
     openid:
       - uid

src/satosa/plugins/microservices/attribute_authorization.yaml

@@ -0,0 +1,9 @@
+module: satosa.micro_services.attribute_authorization.AttributeAuthorization
+name: AttributeAuthorization
+config:
+  force_attributes_presence_on_allow: true
+  attribute_allow:
+    default: # any requester (SP/RP)
+      default: # any issuer (IdP/OP)
+        eduPersonAffiliation:
+          - "^employee$"

src/satosa/proxy_conf.yaml

@@ -12,6 +12,7 @@ FRONTEND_MODULES:
   - plugins/frontends/openid_connect_frontend.yaml
   - plugins/frontends/ping_frontend.yaml
 MICRO_SERVICES:
+  - plugins/microservices/attribute_authorization.yaml
   - plugins/microservices/primary_identifier.yaml
   - plugins/microservices/static_attributes.yaml
   - plugins/microservices/attribute_processor.yaml

src/satosa/tests/test_e2e.py

@@ -2,7 +2,7 @@ import json
 import os
 
 import pytest
-from playwright.sync_api import Browser, BrowserContext, Page, expect
+from playwright.sync_api import Browser, Page, expect
 
 
 @pytest.fixture(scope="session")
@@ -10,10 +10,12 @@ def browser_context_args():
     return {"locale": "fr-FR"}
 
 
-def renater_test_idp(page):
-    page.get_by_label("Nom d'utilisateur").fill("etudiant1")
-    page.get_by_label("Mot de passe").fill("etudiant1")
+def renater_test_idp(page, login):
+    page.get_by_label("Nom d'utilisateur").fill(login)
+    page.get_by_label("Mot de passe").fill(login)
+    page.get_by_label("Afficher les informations qui vont être transférées").check()
     page.get_by_role("button", name="Connexion").click()
+    page.get_by_role("button", name="Accepter").click()
 
 
 def renater_wayf(page):
@@ -22,24 +24,30 @@ def renater_wayf(page):
     page.get_by_role("button", name="Sélection").click()
 
 
-def oidc_to_renater(context: BrowserContext):
-    with context.new_page() as page:
-        page.goto("https://oidc-test-client.traefik.me")
-        renater_wayf(page)
-        renater_test_idp(page)
+def oidc_to_renater(
+    page: Page,
+    login="enseignant1",
+    expected_email="[email protected]",
+    expected_given_name="Georges",
+    expected_usual_name="Grospieds",
+):
+    page.goto("https://oidc-test-client.traefik.me")
+    renater_wayf(page)
+    renater_test_idp(page, login=login)
 
-        expect(page.locator("pre")).to_contain_text('"usual_name":"Dupont"')
-        text = page.inner_text("pre")
-        result = json.loads(text)
+    expect(page.locator("pre")).to_contain_text('"usual_name":')
+    text = page.inner_text("pre")
+    result = json.loads(text)
 
     id_token = result["access_token_response"]["id_token"]
     assert {"acr": "eidas1"}.items() <= id_token.items()
     userinfo = result["userinfo"]
     assert {
-        "email": "jean.dupont@formation.renater.fr",
-        "given_name": "Jean",
-        "uid": "[email protected]",
-        "usual_name": "Dupont",
+        "sub": f"{login}@test-renater.fr",
+        "uid": f"{login}@test-renater.fr",
+        "email": expected_email,
+        "given_name": expected_given_name,
+        "usual_name": expected_usual_name,
     }.items() <= userinfo.items()
     return id_token
 
@@ -47,43 +55,74 @@ def oidc_to_renater(context: BrowserContext):
 @pytest.mark.skipif(
     "TEST_E2E" not in os.environ, reason="Depends on app running locally"
 )
-def test_oidc_to_renater(browser: Browser):
-    id_token1 = oidc_to_renater(browser.new_context())
-    id_token2 = oidc_to_renater(browser.new_context())
+def test_oidc_to_renater_keeps_sub(browser: Browser):
+    with browser.new_context().new_page() as page:
+        id_token1 = oidc_to_renater(page)
+    with browser.new_context().new_page() as page:
+        id_token2 = oidc_to_renater(page)
 
     assert id_token1["sub"] == id_token2["sub"]
 
 
-def agent_connect_login(page: Page):
+@pytest.mark.skipif(
+    "TEST_E2E" not in os.environ, reason="Depends on app running locally"
+)
+def test_oidc_to_renater_student_not_allowed(page: Page):
+    page.goto("https://oidc-test-client.traefik.me")
+    renater_wayf(page)
+    renater_test_idp(page, login="etudiant1")
+
+    expect(page.locator("pre")).to_contain_text('"error":"access_denied"')
+
+
+def agent_connect_login(page: Page, email):
     page.goto("https://fsa1v2.integ01.dev-agentconnect.fr/")
     page.get_by_label("Connexion à AgentConnect").click()
-    page.get_by_label("Email professionnel").fill("[email protected]")
+    page.get_by_label("Email professionnel").fill(email)
     page.get_by_test_id("interaction-connection-button").click()
 
 
-def agent_connect_to_renater(context: BrowserContext):
-    with context.new_page() as page:
-        agent_connect_login(page)
-        renater_wayf(page)
-        renater_test_idp(page)
+def agent_connect_to_renater(
+    page: Page,
+    login="enseignant1",
+    expected_email="[email protected]",
+    expected_given_name="Georges",
+    expected_usual_name="Grospieds",
+):
+    agent_connect_login(page, email=expected_email)
+    renater_wayf(page)
+    renater_test_idp(page, login=login)
 
-        expect(page.locator("body")).to_contain_text("[email protected]")
-        text = page.inner_text("#json")
-        result = json.loads(text)
-        assert {
-            "email": "jean.dupont@formation.renater.fr",
-            "given_name": "Jean",
-            "uid": "[email protected]",
-            "usual_name": "Dupont",
-        }.items() <= result.items()
-        return result
+    expect(page.locator("body")).to_contain_text(expected_email)
+    text = page.inner_text("#json")
+    result = json.loads(text)
+    assert {
+        "uid": f"{login}@test-renater.fr",
+        "email": expected_email,
+        "given_name": expected_given_name,
+        "usual_name": expected_usual_name,
+    }.items() <= result.items()
+    return result
 
 
 @pytest.mark.skipif(
     "TEST_E2E_AC" not in os.environ, reason="Depends on staging deployment"
 )
-def test_agent_connect_to_renater(browser: Browser):
-    result1 = agent_connect_to_renater(browser.new_context())
-    result2 = agent_connect_to_renater(browser.new_context())
+def test_agent_connect_to_renater_keeps_sub(browser: Browser):
+    with browser.new_context().new_page() as page:
+        result1 = agent_connect_to_renater(page)
+    with browser.new_context().new_page() as page:
+        result2 = agent_connect_to_renater(page)
 
     assert result1["sub"] == result2["sub"]
+
+
+@pytest.mark.skipif(
+    "TEST_E2E_AC" not in os.environ, reason="Depends on staging deployment"
+)
+def test_agent_connect_to_renater_student_not_allowed(page: Page):
+    agent_connect_login(page, email="[email protected]")
+    renater_wayf(page)
+    renater_test_idp(page, login="etudiant1")
+
+    expect(page.locator("body")).to_contain_text("access_denied")