@@ -468,3 +468,219 @@ async def _session():
468468
469469 assert resp .status_code == 403
470470 await eng .dispose ()
471+
472+
473+ # --------------------------------------------------------------------------
474+ # Path 2b: Pangolin-Standard-Headers (Remote-User + X-Pangolin-Token)
475+ # --------------------------------------------------------------------------
476+
477+ TRUST_TOKEN = "aabbccddeeff00112233445566778899aabbccddeeff00112233445566778899" # nosec B105 noqa: S105 # ggignore
478+
479+
480+ def _make_sso_app_with_trust (required_scope : str = "read" ):
481+ """Build a minimal app with Pangolin-standard SSO headers configured."""
482+ from app .auth .dependencies import require_scope
483+ from app .config import Settings
484+ from app .db .session import get_session
485+ from fastapi import Depends , FastAPI
486+ from sqlalchemy .ext .asyncio import async_sessionmaker , create_async_engine
487+ from sqlmodel import SQLModel
488+
489+ eng = create_async_engine ("sqlite+aiosqlite:///:memory:" )
490+ settings = Settings (
491+ _env_file = None ,
492+ sso_trust_token = TRUST_TOKEN ,
493+ )
494+ test_app = FastAPI ()
495+
496+ @test_app .get ("/test" )
497+ async def ep (ctx = Depends (require_scope (required_scope , settings = settings ))):
498+ return {"source" : ctx .source , "scope" : ctx .scope }
499+
500+ async def override_session ():
501+ factory = async_sessionmaker (eng , expire_on_commit = False )
502+ async with eng .begin () as conn :
503+ await conn .run_sync (SQLModel .metadata .create_all )
504+ async with factory () as s :
505+ yield s
506+
507+ test_app .dependency_overrides [get_session ] = override_session
508+ return test_app , eng
509+
510+
511+ @pytest .mark .asyncio
512+ async def test_sso_session_with_remote_user_and_trust_token ():
513+ """Remote-User + korrekter X-Pangolin-Token → 200 (Pangolin-Standard)."""
514+ test_app , eng = _make_sso_app_with_trust ("read" )
515+
516+ async with AsyncClient (transport = ASGITransport (app = test_app ), base_url = "http://t" ) as client :
517+ resp = await client .get (
518+ "/test" ,
519+ headers = {
520+ "Remote-User" : "sso-user@example.com" ,
521+ "X-Pangolin-Token" : TRUST_TOKEN ,
522+ },
523+ )
524+
525+ assert resp .status_code == 200 , f"Expected 200, got { resp .status_code } { resp .text }
526+ assert resp .json ()["source" ] == "pangolin-sso"
527+ await eng .dispose ()
528+
529+
530+ @pytest .mark .asyncio
531+ async def test_sso_session_remote_user_without_trust_token ():
532+ """Remote-User ohne X-Pangolin-Token → 401 (nicht vertrauenswürdig)."""
533+ test_app , eng = _make_sso_app_with_trust ("read" )
534+
535+ async with AsyncClient (transport = ASGITransport (app = test_app ), base_url = "http://t" ) as client :
536+ resp = await client .get (
537+ "/test" ,
538+ headers = {"Remote-User" : "evil-user@attacker.com" },
539+ )
540+
541+ assert resp .status_code == 401 , f"Expected 401, got { resp .status_code } { resp .text }
542+ await eng .dispose ()
543+
544+
545+ @pytest .mark .asyncio
546+ async def test_sso_session_trust_token_wrong_value ():
547+ """Remote-User + falscher X-Pangolin-Token → 401 (Token-Mismatch)."""
548+ test_app , eng = _make_sso_app_with_trust ("read" )
549+
550+ async with AsyncClient (transport = ASGITransport (app = test_app ), base_url = "http://t" ) as client :
551+ resp = await client .get (
552+ "/test" ,
553+ headers = {
554+ "Remote-User" : "sso-user@example.com" ,
555+ "X-Pangolin-Token" : "wrong-token-value" ,
556+ },
557+ )
558+
559+ assert resp .status_code == 401 , f"Expected 401, got { resp .status_code } { resp .text }
560+ await eng .dispose ()
561+
562+
563+ @pytest .mark .asyncio
564+ async def test_sso_no_trust_token_configured_rejects_remote_user ():
565+ """Wenn sso_trust_token leer → Remote-User wird NICHT akzeptiert (Sicherheitsstandard)."""
566+ from app .auth .dependencies import require_scope
567+ from app .config import Settings
568+ from app .db .session import get_session
569+ from fastapi import Depends , FastAPI
570+ from sqlalchemy .ext .asyncio import async_sessionmaker , create_async_engine
571+ from sqlmodel import SQLModel
572+
573+ eng = create_async_engine ("sqlite+aiosqlite:///:memory:" )
574+ settings = Settings (_env_file = None , sso_trust_token = "" ) # leer = SSO disabled
575+ test_app = FastAPI ()
576+
577+ @test_app .get ("/test" )
578+ async def ep (ctx = Depends (require_scope ("read" , settings = settings ))):
579+ return {"source" : ctx .source }
580+
581+ async def override_session ():
582+ factory = async_sessionmaker (eng , expire_on_commit = False )
583+ async with eng .begin () as conn :
584+ await conn .run_sync (SQLModel .metadata .create_all )
585+ async with factory () as s :
586+ yield s
587+
588+ test_app .dependency_overrides [get_session ] = override_session
589+
590+ async with AsyncClient (transport = ASGITransport (app = test_app ), base_url = "http://t" ) as client :
591+ resp = await client .get (
592+ "/test" ,
593+ headers = {
594+ "Remote-User" : "sso-user@example.com" ,
595+ "X-Pangolin-Token" : TRUST_TOKEN ,
596+ },
597+ )
598+
599+ assert resp .status_code == 401 , f"Expected 401, got { resp .status_code } { resp .text }
600+ await eng .dispose ()
601+
602+
603+ @pytest .mark .asyncio
604+ async def test_sso_backwards_compat_x_pangolin_user ():
605+ """X-Pangolin-User Header akzeptiert (Rückwärtskompatibilität, kein Trust-Token nötig)."""
606+ from app .auth .dependencies import require_scope
607+ from app .config import Settings
608+ from app .db .session import get_session
609+ from fastapi import Depends , FastAPI
610+ from sqlalchemy .ext .asyncio import async_sessionmaker , create_async_engine
611+ from sqlmodel import SQLModel
612+
613+ eng = create_async_engine ("sqlite+aiosqlite:///:memory:" )
614+ settings = Settings (_env_file = None ) # sso_trust_token leer — X-Pangolin-User trotzdem erlaubt
615+ test_app = FastAPI ()
616+
617+ @test_app .get ("/test" )
618+ async def ep (ctx = Depends (require_scope ("read" , settings = settings ))):
619+ return {"source" : ctx .source }
620+
621+ async def override_session ():
622+ factory = async_sessionmaker (eng , expire_on_commit = False )
623+ async with eng .begin () as conn :
624+ await conn .run_sync (SQLModel .metadata .create_all )
625+ async with factory () as s :
626+ yield s
627+
628+ test_app .dependency_overrides [get_session ] = override_session
629+
630+ async with AsyncClient (transport = ASGITransport (app = test_app ), base_url = "http://t" ) as client :
631+ resp = await client .get (
632+ "/test" ,
633+ headers = {"X-Pangolin-User" : "legacy-user@example.com" },
634+ )
635+
636+ assert resp .status_code == 200 , (
637+ f"Expected 200 for legacy X-Pangolin-User, got { resp .status_code } { resp .text }
638+ )
639+ assert resp .json ()["source" ] == "pangolin-sso"
640+ await eng .dispose ()
641+
642+
643+ @pytest .mark .asyncio
644+ async def test_sso_configurable_header_names ():
645+ """Benutzerdefinierte SSO-Header-Namen werden korrekt ausgewertet."""
646+ from app .auth .dependencies import require_scope
647+ from app .config import Settings
648+ from app .db .session import get_session
649+ from fastapi import Depends , FastAPI
650+ from sqlalchemy .ext .asyncio import async_sessionmaker , create_async_engine
651+ from sqlmodel import SQLModel
652+
653+ eng = create_async_engine ("sqlite+aiosqlite:///:memory:" )
654+ settings = Settings (
655+ _env_file = None ,
656+ sso_user_header = "X-Custom-User" ,
657+ sso_trust_header = "X-Custom-Trust" ,
658+ sso_trust_token = TRUST_TOKEN ,
659+ )
660+ test_app = FastAPI ()
661+
662+ @test_app .get ("/test" )
663+ async def ep (ctx = Depends (require_scope ("read" , settings = settings ))):
664+ return {"source" : ctx .source }
665+
666+ async def override_session ():
667+ factory = async_sessionmaker (eng , expire_on_commit = False )
668+ async with eng .begin () as conn :
669+ await conn .run_sync (SQLModel .metadata .create_all )
670+ async with factory () as s :
671+ yield s
672+
673+ test_app .dependency_overrides [get_session ] = override_session
674+
675+ async with AsyncClient (transport = ASGITransport (app = test_app ), base_url = "http://t" ) as client :
676+ resp = await client .get (
677+ "/test" ,
678+ headers = {
679+ "X-Custom-User" : "custom-user@example.com" ,
680+ "X-Custom-Trust" : TRUST_TOKEN ,
681+ },
682+ )
683+
684+ assert resp .status_code == 200 , f"Expected 200, got { resp .status_code } { resp .text }
685+ assert resp .json ()["source" ] == "pangolin-sso"
686+ await eng .dispose ()
0 commit comments