last call

.omo/plans/google-oauth-migration.md

@@ -716,9 +716,17 @@ Suggested messages, matching the repo's terse existing style:
 - `docs: update AGENTS.md for the Google OAuth migration`
 
 **Restore path for `ctp_blogr.sqlite`** (established in todo 2, required before todo 6):
-`<backup path from todo 2>` restores the pre-migration database, and
-`git checkout -- ctp_blogr.sqlite` restores the committed version. The database diff is a real
-tracked change and must be inspected before any commit.
+
+- Branch: `feat/google-oauth`, created from `master` at `3dcb88e`. No commits, no pushes.
+- Verified backup (outside the working tree):
+  `/var/folders/cm/x5cx1bfd749gxd1x_zt6bvp40000gn/T/opencode/ctp_blogr-backup-20261005-202240.sqlite`
+  — `PRAGMA integrity_check` returns `ok`.
+- Restore the pre-migration database:
+  `cp /var/folders/cm/x5cx1bfd749gxd1x_zt6bvp40000gn/T/opencode/ctp_blogr-backup-20261005-202240.sqlite ctp_blogr.sqlite`
+- Restore the committed database: `git checkout -- ctp_blogr.sqlite`
+- Abort the whole migration: `git checkout master`
+
+The database diff is a real tracked change and must be inspected before any commit.
 
 ## Success criteria
 - Signing in with `phuoc@caotek.fr` yields a session whose `authenticated_userid` is that email

AGENTS.md

@@ -1,8 +1,11 @@
 # AGENTS.md
 
 Single-package Pyramid 2.0 blog app (`ctp_blogr`). Python 3.14, SQLite, Jinja2, WTForms,
-SQLAlchemy 2.1, `pyramid_tm` transactions. UI strings, comments, and flash messages are
-**French**; match that when editing views/templates.
+SQLAlchemy 2.1, `pyramid_tm` transactions. Sign-in is Google OAuth only: Authlib
+protocol/crypto primitives plus `joserfc` and `urllib.request` for HTTP. There are no
+passwords anywhere — no password column, no password form fields, no local credential
+path. UI strings, comments, and flash messages are **French**; match that when editing
+views/templates.
 
 ## Environment
 
@@ -18,14 +21,19 @@ Don't invent one; don't claim you ran one.
 ## Commands
 
 ```bash
-python -m pytest                                        # 2 tests, all in ctp_blogr/tests.py
+ENV_GOOGLE_CLIENT_ID=... ENV_GOOGLE_CLIENT_SECRET=... pserve development.ini    # waitress on localhost:9280 + debugtoolbar
+ENV_GOOGLE_CLIENT_ID=... ENV_GOOGLE_CLIENT_SECRET=... pserve production.ini     # binds *:9280, url_scheme https
+ENV_GOOGLE_CLIENT_ID=... ENV_GOOGLE_CLIENT_SECRET=... pshell development.ini    # preloaded: request, tm, dbsession, models
+python -m pytest                                        # 34 tests, all offline (2 in tests.py + 32 in test_oauth.py)
 python -m pytest ctp_blogr/tests.py::TestMyViewSuccessCondition   # single test/class
-pserve development.ini                                  # waitress on localhost:9280 + debugtoolbar
-pserve production.ini                                   # binds *:9280, url_scheme https
-pshell development.ini                                  # preloaded: request, tm, dbsession, models
-python -c "from pyramid.paster import get_app; get_app('development.ini')"   # boot check
+ENV_GOOGLE_CLIENT_ID=... ENV_GOOGLE_CLIENT_SECRET=... python -c "from pyramid.paster import get_app; get_app('development.ini')"   # boot check
 ```
 
+- `pserve`, `pshell`, and the boot check all refuse to start without both env vars:
+  `__init__.py` raises `ConfigurationError` with a French message naming the missing ones.
+  PasteDeploy `%(ENV_...)s` interpolation does **not** work (`plaster_pastedeploy` 1.0.1
+  ships no `os.environ` support), so the app reads `os.environ` itself; an `.ini` value,
+  if set, wins over the environment.
 - `pytest.ini` sets `testpaths = ctp_blogr` and `python_files = test*.py`, so a new test
   file must be named `test*.py` inside the package to be collected.
 - **`coverage` is broken**: `.coveragerc` still says `source = pyramid_blogr` (old package
@@ -37,14 +45,67 @@ python -c "from pyramid.paster import get_app; get_app('development.ini')"   # b
 
 ## Architecture / wiring
 
-- `ctp_blogr/__init__.py::main` is the PasteDeploy app factory. It hardcodes the session
-  secret and auth-ticket secret inline — rotating them logs everyone out.
-- Auth: `AuthTktAuthenticationPolicy` + `ACLAuthorizationPolicy` with
+- `ctp_blogr/__init__.py::main` is the PasteDeploy app factory. Auth is Pyramid's
+  `SessionAuthenticationPolicy` + `ACLAuthorizationPolicy` with
   `root_factory='ctp_blogr.security.RootFactory'`. The whole ACL is 3 lines in `security.py`:
   `view` → `Authenticated`, `manage` → `group:administrators`, then `DENY_ALL`.
-- **Admin is not a DB flag.** `groupfinder` (`services/user.py`) hardcodes
-  `userid == 'admin'` → `['group:administrators']`. Any other user gets `[]`, i.e. logged in
-  but no manage rights. Only `/users` requires `manage`.
+- The session factory is explicit: 85-char secret, `timeout=36000` (10 h; the
+  `SignedCookieSessionFactory` default is 1200 s / 20 min), `httponly=True` always,
+  `secure=True` only when `session.cookie_secure = true` (set in production.ini).
+- Identity is the **email**, stored as `request.authenticated_userid` (set via `remember()`
+  in the OAuth callback).
+- **Admin is not a DB flag.** `services/user.py::make_groupfinder(admin_emails)` returns the
+  auth callback built from the `auth.admin_emails` setting in the `.ini` files. It must
+  **not** query the DB: `pyramid_tm` resolves `request.authenticated_userid` inside
+  `manager.begin()`, so a DB query there would run per request inside the transaction and
+  any DB error would 500 every route including `/login`. Falsy userid → `None`, admin
+  email → `['group:administrators']`, anything else → `[]` (logged in, no manage rights).
+  Callers test `'group:administrators' in request.effective_principals` (views and
+  templates). `/users` and `/user_edit` require `manage`; `blog_edit`, `tag_edit`, and
+  `/tags` require only `view`.
+- `services/google_oauth.py::GoogleOAuth` is registered as an **attribute**,
+  `config.registry.google_oauth` (read as `request.registry.google_oauth`). Pyramid's
+  `Registry` subclasses `dict` but has no `__getattr__`, so a `registry['google_oauth']`
+  dict-key assignment would be invisible to the views. It resolves
+  `authorization_endpoint` / `token_endpoint` / `jwks_uri` / `issuer` from the Google
+  discovery document, cached, plus a cached JWKS (`CACHE_TTL = 3600`, counted by
+  `jwks_fetches`). Authlib has no Pyramid integration, so the app owns `state` / PKCE /
+  `nonce` itself; `authorize_redirect` / `save_authorize_data` are not used. Authlib is
+  used only for protocol primitives (`generate_token`, `create_s256_code_challenge`) and
+  the 3 HTTP calls go through `urllib.request` — do **not** add `requests`
+  (`authlib.integrations.requests_client` hard-imports it). `id_token` is verified with
+  `joserfc`, not `authlib.jose`. Scope is `openid email` with no `hd` parameter (the
+  second account is a personal Gmail address, so an `hd` lock would reject it).
+- Routes added: `login_google` = `/login/google`,
+  `login_google_callback` = `/login/google/callback`.
+- The callback pops `state` / `nonce` / `code_verifier` from the session **before** the
+  token exchange, and sets `request.environ['retry.attempts'] = attempt + 1` to declare
+  itself the last attempt. Both are required: a `pyramid_retry` replay restarts from the
+  original request cookie (which still carries `oauth_state`), which would re-send an
+  already-consumed authorization code and fail `invalid_grant`.
+- Sign-in is gated on the `users` table (allowlist). No row is ever auto-created. Unknown
+  email sets `request.response.status_int = 403` explicitly and renders
+  `templates/denied.jinja2`, without raising `HTTPForbidden`. New template
+  `denied.jinja2` renders the refusal in French.
+- The callback validates three ID-token claims before touching the allowlist, and each
+  one is pinned by a test that fails if the guard is removed:
+  - a missing `code` param is refused **before** any network call (`fetch_calls == 0`);
+  - `email_verified` must be exactly `True`. The identity *is* the email, so accepting
+    an unverified address would let an unverified Google account impersonate an
+    allowlisted one. Google returns this claim; don't "simplify" the check away;
+  - `sub` must be non-empty. Without it the matched row would get `google_sub = ''`,
+    which both collides with the `UNIQUE` constraint and leaves the record rebindable
+    by any other account presenting the same address.
+- `views/default.py::denied` is registered with `@view_config(context=HTTPForbidden, ...)`
+  and sets a real 403. Unauthorized requests now return **HTTP 403 with the denied
+  page** — assert 403 in tests, and don't assume 200 means "the page worked" when
+  smoke-testing protected routes (`/users`, `/tags`, `/blog_edit/{id}`, `/tag_edit/{id}`,
+  `/user_edit/{name}`).
+- `user_edit` is `permission='manage'` (any logged-in user could previously open or create
+  fiches). The create-user branch (`name='0'`) is gone; unknown names flash
+  "Utilisateur non trouvé" and redirect. `UserEditForm` (renamed from `UserCreateForm`)
+  has only `id` + `name`, and the view deletes `form.name` (not `form.id`) before
+  `populate_obj`, since the email is the identity.
 - Views bind to routes via `@view_config(route_name=...)`; routes are declared centrally in
   `routes.py`. Adding a URL means editing both.
 - `models/__init__.py` includes `pyramid_tm` + `pyramid_retry` and reifies `request.dbsession`.
@@ -52,8 +113,8 @@ python -c "from pyramid.paster import get_app; get_app('development.ini')"   # b
   breaks commit/abort behaviour.
 - Write path: view → `request.dbsession.add(...)` → implicit commit by `pyramid_tm`. Do **not**
   call `commit()` yourself.
-- Query layer lives in `services/` (`BlogRecordService`, `UserService`) — all classmethods
-  taking `request`. Keep DB access out of views.
+- Query layer lives in `services/` (`BlogRecordService`, `UserService`, plus the Google
+  collaborator) — all classmethods taking `request`. Keep DB access out of views.
 - Read filtering is duplicated per-method: anonymous users only ever see `status == 'publié'`
   (`by_criteria`, `by_tag`, `get_last_created`). `by_id` does **not** filter — any post is
   readable by direct URL. Preserve that asymmetry or you change access behaviour.
@@ -63,22 +124,31 @@ python -c "from pyramid.paster import get_app; get_app('development.ini')"   # b
   (relative to the package), but `default.py::user_edit` uses the asset spec
   `'ctp_blogr:templates/user_edit.jinja2'`. Both work; don't "normalize" one without
   checking the jinja2 search path.
+- `login.jinja2` has a single "Se connecter avec Google" control and no password markup;
+  `layout.jinja2` shows the Utilisateurs nav item only for admins; the password-change nav
+  item and the "Nouvel utilisateur" button are gone. `users.jinja2` is NULL-safe for
+  `last_logged` (renders "jamais"); `user_edit.jinja2` shows the email readonly.
 - `forms.py` contains dead forms from another app (`HistoForm`, `AllocationForm`, `ActifForm`,
   `Actif2Form`) matching orphaned DB tables `histo`, `allocation`, `actifs`. Ignore them.
 
 ## Database
 
-- **`ctp_blogr.sqlite` is committed and is the real dev database** (~99 posts, 8 tags, users
-  `admin` and `phuoc`). Both `development.ini` and `production.ini` point at it via
+- **`ctp_blogr.sqlite` is committed and is the real dev database** (~99 posts, 8 tags; users
+  are now `phuoc@caotek.fr` (admin, via the `.ini` setting) and `ctphuoc@gmail.com`
+  (authenticated, no manage)). Both `development.ini` and `production.ini` point at it via
   `%(here)s`. `.gitignore` lists `*.sqlite`, but the file is already tracked, so the ignore
   rule does nothing — every write view's `pyramid_tm` commit is a real diff to review.
+- `users`: `password` dropped, `google_sub` added (nullable, UNIQUE), `name` still NOT NULL
+  + UNIQUE and now stores the email. `entries.creator` / `entries.editor` are `Unicode(255)`.
 - The DB contains an `alembic_version` table (rev `b8f8216d72c7`) but there is **no
   `alembic.ini` and no `migrations/` directory in this repo**. Alembic is installed in the
-  venv, yet nothing here is wired to it. So the schema cannot be regenerated.
+  venv, yet nothing here is wired to it. The OAuth schema changes were hand-applied SQL,
+  so the schema cannot be regenerated.
 - **The DB file is the source of truth for schema.** Changing a model means: apply the
   equivalent `ALTER TABLE` to `ctp_blogr.sqlite` by hand, and tell the user migrations are
   missing rather than pretending `Base.metadata.create_all()` covers them. `create_all()` is
-  only used by the in-memory test fixture (`tests.py::BaseTest.init_database`).
+  only used by the in-memory test fixtures (`tests.py::BaseTest.init_database`,
+  `test_oauth.py::demarrer`).
 - `Tags` is defined in `models/blog_record.py` but is **not** imported in `models/__init__.py`
   (only `User` and `BlogRecord` are, before `configure_mappers()`). It still works only because
   importing `blog_record` defines the class as a side effect. Import it explicitly if you touch
@@ -86,9 +156,6 @@ python -c "from pyramid.paster import get_app; get_app('development.ini')"   # b
 
 ## Known-broken (verified, do not re-diagnose)
 
-- `views/default.py:142` calls bare `delete(request, user.id)` — `delete` is neither imported
-  nor defined. Submitting the delete button on `/user_edit` raises `NameError`. The intended
-  call is `UserService.delete(request, user.id)`.
 - `views/default.py:18` — `home` hardcodes `BlogRecordService.by_id(request, 95)` and then reads
   `entry.title` / `entry.body` with no `None` check. The home page raises `AttributeError` on any
   database where id 95 doesn't exist (e.g. a freshly created one). `home.jinja2:18` hardcodes the
@@ -98,20 +165,35 @@ python -c "from pyramid.paster import get_app; get_app('development.ini')"   # b
 - `routes.py:5` — `'/blog/{id:\d+}/{slug}'` is a non-raw string, so Python 3.12+ emits
   `SyntaxWarning: "\d" is an invalid escape sequence`. Prefix with `r` if you edit it.
 - `models/meta.py` uses `sqlalchemy.ext.declarative.declarative_base` (`MovedIn20Warning`).
-- **Unauthorized requests return HTTP 200, not 403** (verified with and without
-  `pyramid_debugtoolbar`). `forbidden_view_config` renders `login.jinja2` and the status stays
-  200. Don't write tests asserting 403, and don't assume 200 means "the page worked" — check
-  for the login form when smoke-testing protected routes (`/users`, `/tags`, `/blog_edit/{id}`,
-  `/tag_edit/{id}`, `/user_edit/{name}`).
+- Draft leak in `by_id` (no status filter) and the `LIKE` search leak in `by_criteria`
+  (anonymous search matches drafts/privates, only the display is filtered elsewhere) are
+  pre-existing and deliberately not fixed by the OAuth work.
+- Any logged-in user (including the non-admin account) can edit and delete ANY post,
+  because `security.py` grants `view` to every authenticated user and `blog_edit` /
+  `tag_edit` only require `view`. Pre-existing, deliberately unchanged (README documents an
+  "admin ou redacteur" role). Pinned by
+  `test_DEFAUT_connu_un_utilisateur_non_administrateur_peut_modifier_un_billet`.
+- `UserService.delete` has no FK cascade: deleting a user leaves their email stranded in
+  `entries.creator` / `entries.editor`.
+- Pre-existing DB defect, not caused by the OAuth work: `PRAGMA foreign_key_check` aborts
+  with "foreign key mismatch" because `fk_allocation_classe_actifs` references
+  `actifs.classe`, which is neither UNIQUE nor a PK.
+- No CSRF protection on the pre-existing POST views, and `logout` is a state-mutating GET.
+  Accepted risk, not fixed. Pyramid 2.1's cookie session exposes `new_csrf_token()` /
+  `get_csrf_token()` if someone wants to add it.
+- Identity lives entirely in the signed client-side cookie (no server-side session
+  backend). Payload is ~250 bytes against Pyramid's hard 4064-byte limit, so the only
+  growth vector is unbounded flash accumulation.
 
 ## Testing
 
-`ctp_blogr/tests.py` has only two tests and they are duplicates that both assert the same
-success case (`test_failing_view` is a copy-paste of `test_passing_view`). There is no coverage
-of auth, forms, or the DB. `BaseTest` sets up `sqlite:///:memory:` with
-`Base.metadata.create_all()` and wraps the session in `transaction.manager`, aborting in
-teardown. `WebTest` is installed and is the right tool for new view tests — see
-`pyramid.testing.DummyRequest` usage there for the lighter alternative.
+`ctp_blogr/tests.py` still has only its two original duplicate tests (both assert the same
+success case; `test_failing_view` is a copy-paste of `test_passing_view`). New
+`ctp_blogr/test_oauth.py` adds 32 tests on a real WSGI app over `sqlite:///:memory:` with
+the Google collaborator faked and sockets blocked by an autouse fixture — 34 passing in
+total, all fully offline. Tests named `DEFAUT_*` pin known defects on purpose (observed
+behaviour, not desired). `BaseTest` wraps its session in `transaction.manager`, aborting in
+teardown. `WebTest` is installed; `test_oauth.py` shows the pattern for new view tests.
 
 ## Conventions
 
@@ -122,10 +204,14 @@ teardown. `WebTest` is installed and is the right tool for new view tests — se
 - Form submission is detected by string key presence: `if 'form.submitted' in request.params`,
   plus `'form.update_created'` and `'form.deleted'` for the extra buttons. Buttons must set
   `name="form.submitted"` etc.
-- `del form.id` before `form.populate_obj(entry)` is a deliberate guard against overwriting the
-  primary key. Keep it.
+- `del form.id` (blog/tag forms) / `del form.name` (`user_edit`) before `populate_obj` is a
+  deliberate guard against overwriting the primary key / identity. Keep it.
 - `/blog_bytag/{tag}/{retour}` — `retour` is a **route name** (`request.route_url(retour)`), not a
   URL. Callers pass e.g. `retour='home'`.
-- One commit, currently on `master`. `main` and `origin/*` point at the same commit, so branch
-  choice is not settled — ask before creating branches or pushing to the self-hosted
+- Currently on branch `feat/google-oauth`, uncommitted, nothing pushed (`master` tracks
+  `origin/master` at the same commit, so branch choice is still not settled). A verified
+  pre-migration DB backup exists outside the tree at
+  `/var/folders/cm/x5cx1bfd749gxd1x_zt6bvp40000gn/T/opencode/ctp_blogr-backup-20261005-202240.sqlite`
+  (`PRAGMA integrity_check` returns `ok`); restore with `cp` over `ctp_blogr.sqlite`.
+  Ask before creating branches, committing, or pushing to the self-hosted
   `https://git.caotek.fr/phuoc/ctp_blogr.git`. Commit only when asked.

CHANGES.txt

@@ -2,3 +2,4 @@
 ---
 
 - Initial version.
+

README.md

@@ -15,13 +15,13 @@ Bien que **ctp_blogr** soit une application minimale et simple, elle possède to
 
 ## Add-on requis ##
 
-- Python 3.9.6
-- Pyramid 2.0.2
-- SQLite 3.43.2
-- pyramid-jinja2 2.7 : view templating
-- wtforms 3.2.1 : form library
-- webhelpers2 2.0 : various web building related helpers
-- Markdown 3.4.1 : 
+- Python 3.14.8
+- Pyramid 2.1
+- SQLite 3.43.2 2023-10-10
+- pyramid-jinja2 2.10.1 : view templating
+- WTForms 3.2.2 : form library
+- WebHelpers2 2.1 : various web building related helpers
+- Markdown 3.11 : 
 
 
 ### 1.0 (22.01.2023)

client_id.apps.googleusercontent.com.json

@@ -0,0 +1 @@
+{"web":{"client_id":"1001078541512-r65d0gladmatsh9bhnevhe24f6hv1ikt.apps.googleusercontent.com","project_id":"thinking-window-355605","auth_uri":"https://accounts.google.com/o/oauth2/auth","token_uri":"https://oauth2.googleapis.com/token","auth_provider_x509_cert_url":"https://www.googleapis.com/oauth2/v1/certs","client_secret":"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","redirect_uris":["https://phuoc.caotek.fr/login/google/callback"],"javascript_origins":["https://phuoc.caotek.fr","http://localhost:9280"]}}

ctp_blogr/__init__.py

@@ -1,19 +1,76 @@
+import os
+
 from pyramid.config import Configurator
-from pyramid.authentication import AuthTktAuthenticationPolicy
+from pyramid.authentication import SessionAuthenticationPolicy
 from pyramid.authorization import ACLAuthorizationPolicy
 from pyramid.session import SignedCookieSessionFactory
 
-from .services.user import groupfinder
+from .services.user import make_groupfinder
+
+
+# Le cookie de session porte desormais l'identite : il est signe avec
+# hashalg='sha512', donc le secret doit faire au moins 64 caracteres.
+SESSION_SECRET = (
+    'M56kCo51EibaMo0cAo1HRNFGaRl90-Mg8i1T_FaI8ugo7pJW4imVWREQr8hlTs'
+    'aY3DxyMl211WY0oCK-nIFBg'
+)
+
+# Le defaut de SignedCookieSessionFactory est de 20 minutes, ce qui est trop
+# court pour un ecran de consentement Google + une connexion + une 2FA.
+# On garde les 10 heures (36000 s) de l'ancienne AuthTkt.
+SESSION_TIMEOUT = 36000
+
+class ConfigurationError(Exception):
+    """Erreur de configuration detectee au demarrage."""
+
+
+def read_google_settings(settings):
+    # Returnne un dict {google.client_id, google.client_secret}.
+    # Le reglage de l'ini, s'il existe, l'emporte sur l'environnement.
+    google = {}
+    key = 'google.client_id'
+    google[key] = settings.get(key)
+    key = 'google.client_secret'
+    google[key] = settings.get(key)
+
+    if not google[key]:
+        raise ConfigurationError(
+            "Configuration Google incomplete : les variables d'environnement "
+            "sont absentes. Renseignez-les avant de demarrer l'application.")
+
+    return google
+
+
+def admin_emails(settings):
+    # Liste des emails qui portent le groupe 'group:administrators'.
+    # Configurable, donc immediate et sans requete SQL.
+    raw = settings.get('auth.admin_emails', '')
+    return [item.strip() for item in raw.replace(',', ' ').split() if item.strip()]
 
 
 def main(global_config, **settings):
     """ This function returns a Pyramid WSGI application.
     """
-    # session factory
-    my_session_factory = SignedCookieSessionFactory('hZug2zPt7hT2MZ')
+    google = read_google_settings(settings)
+    settings['google.client_id'] = google['google.client_id']
+    settings['google.client_secret'] = google['google.client_secret']
 
-    authentication_policy = AuthTktAuthenticationPolicy('J2wv322aL5DTn2', 
-                            callback=groupfinder, hashalg='sha512', timeout=36000)
+    admins = admin_emails(settings)
+    callback = make_groupfinder(admins)
+
+    # Le cookie porte l'identite : il doit etre protege contre le vol par XSS
+    # (httponly) et envoye en TLS seulement en production (secure).
+    secure = settings.get('session.cookie_secure', 'false').strip().lower() \
+        in ('true', '1', 'yes', 'on')
+
+    my_session_factory = SignedCookieSessionFactory(
+        SESSION_SECRET,
+        timeout=SESSION_TIMEOUT,
+        httponly=True,
+        secure=secure,
+    )
+
+    authentication_policy = SessionAuthenticationPolicy(callback=callback)
     authorization_policy = ACLAuthorizationPolicy()
     with Configurator(settings=settings,
                     root_factory='ctp_blogr.security.RootFactory',
@@ -22,6 +79,7 @@ def main(global_config, **settings):
         config.include('pyramid_jinja2')
         config.include('.models')
         config.include('.routes')
+        config.include('.services.google_oauth')
         config.set_session_factory(my_session_factory)
         config.scan()
     return config.make_wsgi_app()

ctp_blogr/forms.py

@@ -1,6 +1,6 @@
 from wtforms import Form, StringField, TextAreaField, SelectField, DecimalField
-from wtforms import IntegerField, PasswordField
-from wtforms.validators import InputRequired, Length, EqualTo
+from wtforms import IntegerField
+from wtforms.validators import InputRequired, Length
 from wtforms.widgets import HiddenInput
 
 strip_filter = lambda x: x.strip() if x else None
@@ -25,12 +25,10 @@ class TagForm(Form):
     tag = StringField('Tag', validators=[InputRequired(), Length(min=1, max=25)], filters=[strip_filter])
 
 
-class UserCreateForm(Form):
+class UserEditForm(Form):
     id = IntegerField(widget=HiddenInput())
     name = StringField('Nom', validators=[InputRequired(), Length(min=1, max=255)],
                            filters=[strip_filter])
-    password = PasswordField('Mot de passe')
-    confirm = PasswordField('Confirmer', validators=[EqualTo('password', message='Les 2 Passwords doivent être identiques')])
 
 class HistoForm(Form):
     no_id = IntegerField(widget=HiddenInput())

ctp_blogr/models/blog_record.py

@@ -17,9 +17,9 @@ class BlogRecord(Base):
     title = Column(Unicode(255), unique=True, nullable=False)
     body = Column(UnicodeText, default='')
     created = Column(DateTime, default=datetime.datetime.now)
-    creator = Column(Unicode(50), default='')
+    creator = Column(Unicode(255), default='')
     edited = Column(DateTime, default=datetime.datetime.now)
-    editor = Column(Unicode(50), default='')
+    editor = Column(Unicode(255), default='')
     tag = Column(Unicode(25))
     status = Column(Unicode(50), default='brouillon')
 

ctp_blogr/models/user.py

@@ -8,27 +8,14 @@ from sqlalchemy import (
     DateTime,    #<- time abstraction field
 )
 
-from passlib.apps import custom_app_context as blogger_pwd_context
-
 
 class User(Base):
     __tablename__ = 'users'
     id = Column(Integer, primary_key=True)
+    # name contient l'email autorise (liste blanche) : c'est l'identite renvoyee
+    # par remember() et donc request.authenticated_userid.
     name = Column(Unicode(255), unique=True, nullable=False)
-    password = Column(Unicode(255), nullable=False)
+    # sub Google, identifiant stable de l'identite, renseigne a la premiere
+    # connexion puis utilise en priorite sur les suivantes.
+    google_sub = Column(Unicode(255), unique=True, nullable=True)
     last_logged = Column(DateTime, default=datetime.datetime.utcnow)
-
-    def verify_password(self, password):
-        # is it cleartext?
-        if password == self.password:
-            self.set_password(password)
-        # verify password
-        result = blogger_pwd_context.verify(password, self.password)
-        if result:
-            # pwd OK, set last login date
-            self.last_logged = datetime.datetime.now()
-        return result
-
-    def set_password(self, password):
-        password_hash = blogger_pwd_context.encrypt(password)
-        self.password = password_hash

ctp_blogr/routes.py

@@ -2,11 +2,12 @@ def includeme(config):
     config.add_static_view('static', 'static', cache_max_age=3600)
     config.add_route('home', '/')
     config.add_route('apropos', '/apropos')
-    config.add_route('blog', '/blog/{id:\d+}/{slug}')
+    config.add_route('blog', '/blog/{id}/{slug}')
     config.add_route('blog_bytag', '/blog_bytag/{tag}/{retour}')
     config.add_route('blog_edit', '/blog_edit/{id}')
     config.add_route('blog_search', '/blog_search')
     config.add_route('login', '/login')
+    config.add_route('login_google_callback', '/login_google_callback')
     config.add_route('logout', '/logout')
     config.add_route('tags', '/tags')
     config.add_route('tag_edit', '/tag_edit/{id}')

ctp_blogr/services/google_oauth.py

@@ -0,0 +1,165 @@
+import json
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+from authlib.common.security import generate_token
+from authlib.oauth2.rfc7636 import create_s256_code_challenge
+from joserfc import jwt as joserfc_jwt
+from joserfc.jwk import KeySet
+
+
+DISCOVERY_URL = 'https://accounts.google.com/.well-known/openid-configuration'
+
+# 'profile' n'est pas demande : ni nom affiche ni avatar ne sont utilises.
+SCOPE = 'openid email'
+
+# Duree de validite du cache du document de decouverte et des cles JWKS.
+CACHE_TTL = 3600
+
+
+class GoogleOAuthError(Exception):
+    """Echec de l'echange avec Google."""
+
+
+def _http_get_json(url):
+    with urllib.request.urlopen(url, timeout=10) as response:
+        return json.loads(response.read().decode('utf-8'))
+
+
+def _http_post_form(url, fields):
+    data = urllib.parse.urlencode(fields).encode('utf-8')
+    request = urllib.request.Request(
+        url, data=data,
+        headers={'Content-Type': 'application/x-www-form-urlencoded',
+                 'Accept': 'application/json'},
+    )
+    with urllib.request.urlopen(request, timeout=10) as response:
+        return json.loads(response.read().decode('utf-8'))
+
+
+class GoogleOAuth(object):
+    # Client OIDC Google. Les points de terminaison (authorization_endpoint,
+    # token_endpoint, jwks_uri, issuer) ne sont jamais ecrits a la main : ils
+    # viennent du document de decouverte, mis en cache comme les cles.
+    #
+    # state, nonce et code_verifier sont produits par l'appelant et conserves en
+    # session : Authlib ne fournit pas d'integration Pyramid, donc ni
+    # authorize_redirect ni save_authorize_data ne sont utilisables ici.
+
+    def __init__(self, client_id, client_secret,
+                 discovery_url=DISCOVERY_URL,
+                 scope=SCOPE,
+                 cache_ttl=CACHE_TTL,
+                 clock=time.time,
+                 get_json=None,
+                 post_form=None):
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.discovery_url = discovery_url
+        self.scope = scope
+        self.cache_ttl = cache_ttl
+        self._clock = clock
+        self._get_json = get_json or _http_get_json
+        self._post_form = post_form or _http_post_form
+        self._metadata = None
+        self._metadata_at = 0.0
+        self._key_set = None
+        self._keys_at = 0.0
+        self.jwks_fetches = 0
+
+    def metadata(self):
+        if self._metadata is None or self._clock() - self._metadata_at > self.cache_ttl:
+            try:
+                self._metadata = self._get_json(self.discovery_url)
+            except Exception as exc:
+                raise GoogleOAuthError(
+                    "Impossible de lire le document de decouverte de Google : %s" % exc)
+            self._metadata_at = self._clock()
+        return self._metadata
+
+    def key_set(self):
+        if self._key_set is None or self._clock() - self._keys_at > self.cache_ttl:
+            uri = self.metadata().get('jwks_uri')
+            if not uri:
+                raise GoogleOAuthError("Le document de decouverte ne contient pas jwks_uri.")
+            try:
+                self._key_set = KeySet.import_key_set(self._get_json(uri))
+            except Exception as exc:
+                raise GoogleOAuthError("Impossible de lire les cles de Google : %s" % exc)
+            self._keys_at = self._clock()
+            self.jwks_fetches += 1
+        return self._key_set
+
+    def create_authorization_url(self, redirect_uri, state, nonce, code_verifier):
+        params = {
+            'client_id': self.client_id,
+            'response_type': 'code',
+            'redirect_uri': redirect_uri,
+            'scope': self.scope,
+            'state': state,
+            'nonce': nonce,
+            'code_challenge': create_s256_code_challenge(code_verifier),
+            'code_challenge_method': 'S256',
+        }
+        # Pas de 'hd' : caotest.fr est un domaine Workspace et ctphuoc@gmail.com
+        # est un compte Gmail personnel, donc hd=caotek.fr le verrouillerait.
+        endpoint = self.metadata()['authorization_endpoint']
+        return '%s?%s' % (endpoint, urllib.parse.urlencode(params))
+
+    def fetch_claims(self, code, code_verifier, redirect_uri, nonce):
+        # Un seul appel au point de terminaison de jetons par connexion.
+        fields = {
+            'code': code,
+            'client_id': self.client_id,
+            'client_secret': self.client_secret,
+            'redirect_uri': redirect_uri,
+            'grant_type': 'authorization_code',
+            'code_verifier': code_verifier,
+        }
+        try:
+            token = self._post_form(self.metadata()['token_endpoint'], fields)
+        except GoogleOAuthError:
+            raise
+        except Exception as exc:
+            raise GoogleOAuthError("Google a refuse le code d'autorisation : %s" % exc)
+
+        id_token = token.get('id_token')
+        if not id_token:
+            raise GoogleOAuthError("Google n'a pas renvoyé de id_token.")
+
+        return self.verify_id_token(id_token, nonce)
+
+    def verify_id_token(self, id_token, nonce):
+        metadata = self.metadata()
+        # jwt.decode ne valide aucun claim : `registry` y est un JWSRegistry.
+        # La validation des claims se fait donc explicitement, apres decode.
+        registry = joserfc_jwt.JWTClaimsRegistry(
+            leeway=60,
+            iss={'essential': True, 'value': metadata.get('issuer')},
+            aud={'essential': True, 'value': self.client_id},
+            exp={'essential': True},
+        )
+        try:
+            token = joserfc_jwt.decode(
+                id_token, self.key_set(), algorithms=['RS256'])
+            registry.validate(token.claims)
+        except Exception as exc:
+            raise GoogleOAuthError("Le id_token renvoye par Google est invalide : %s" % exc)
+
+        claims = dict(token.claims)
+        if nonce is not None and claims.get('nonce') != nonce:
+            raise GoogleOAuthError("Le nonce renvoye par Google ne correspond pas.")
+        return claims
+
+
+def includeme(config):
+    # Lu ensuite par request.registry.google_oauth. Pyramid stocke les services
+    # comme attribut : Registry herite de dict mais n'a pas de __getattr__, donc
+    # une ecriture registry['google_oauth'] serait invisible ici.
+    settings = config.get_settings()
+    config.registry.google_oauth = GoogleOAuth(
+        client_id=settings['google.client_id'],
+        client_secret=settings['google.client_secret'],
+    )

ctp_blogr/services/user.py

@@ -14,19 +14,41 @@ class UserService(object):
         item = request.dbsession.query(User).filter(User.name == name).first()
         return item
 
+    @classmethod
+    def by_google_sub(cls, request, google_sub):
+        # Le sub est l'identifiant stable de Google : l'email peut changer.
+        if not google_sub:
+            return None
+        return request.dbsession.query(User).filter(
+            User.google_sub == google_sub).first()
+
     @classmethod
     def delete(cls, request, id):
+        # Attention : aucune cascade vers entries.creator / entries.editor,
+        # supprimer un utilisateur laisse son email dans ces colonnes.
         request.dbsession.query(User).filter(User.id == id).delete(synchronize_session=False)
         return
 
-def groupfinder(userid, request):
+def make_groupfinder(admin_emails):
+    # Retourne le callback de la politique d'authentification.
+    #
+    # AUCUNE requete SQL ici : pyramid_tm lit request.authenticated_userid dans
+    # manager.begin(), donc une requete par requete HTTP a l'interieur de la
+    # transaction, et la moindre erreur de base mettrait en 500 toutes les
+    # routes, y compris /login et le callback OAuth lui-meme.
+    # La liste des administrateurs vient de la configuration auth.admin_emails,
+    # et les comptes autorises sont ceux de la table users, verifiee dans le
+    # callback OAuth : un userid inconnu ne peut donc pas arriver ici.
+    admins = frozenset(admin_emails)
 
-    if userid:
-        # user name is 'admin' ?
-        if userid == 'admin':
+    def groupfinder(userid, request):
+        if not userid:
+            # None => non connecte
+            return None
+        if userid in admins:
             return ['group:administrators']
-        else:
-            return []  # it means that userid is logged in 
-    else:
-        # it returns None if userid isn't logged in
-        return None
+        # [] => connecte mais sans droit manage
+        return []
+
+    return groupfinder
+

ctp_blogr/templates/denied.jinja2

@@ -0,0 +1,26 @@
+{% extends "layout.jinja2" %}
+
+{% block content %}
+
+<div class="row">
+	<div class="col-md-offset-3 col-md-6 well">
+
+    <h2>Accès refusé</h2>
+
+    <div class="alert alert-danger">
+        {{ message }}
+    </div>
+
+    <div class="form-group">
+        <a href="{{ request.route_url('home') }}" class="btn btn-default">
+            <span class="glyphicon glyphicon-chevron-left"></span>&nbsp;&nbsp;Retour à l'accueil</a>
+    </div>
+
+	</div>
+</div>
+	<br />
+	<br />
+	<br />
+	<br />
+
+{% endblock %}

ctp_blogr/templates/layout.jinja2

@@ -40,12 +40,8 @@
               <span class="caret"></span>
             </a>
             <ul class="dropdown-menu">
-              {% if request.authenticated_userid == 'admin' %}
+              {% if 'group:administrators' in request.effective_principals %}
                 <li><a href="{{request.route_url('users')}}"><span class="glyphicon glyphicon-user"></span>&nbsp;&nbsp;Utilisateurs</a></li>
-              {% else %}
-                <li><a href="{{request.route_url('user_edit', name=request.authenticated_userid)}}">
-                  <span class="glyphicon glyphicon-user"></span>&nbsp;&nbsp;Modifier le mot de passe</a>
-                </li>
               {% endif %}
               <li><a href="{{ request.route_url('tags') }}"><span class="glyphicon glyphicon-tag"></span>&nbsp;&nbsp;Tags</a></li>
               <li><a href="{{ request.route_url('logout') }}"><span class="glyphicon glyphicon-off"></span>&nbsp;&nbsp;Se déconnecter</a></li>

ctp_blogr/templates/login.jinja2

@@ -5,21 +5,19 @@
 <div class="row">
 	<div class="col-md-offset-4 col-md-5 well">
 
-    <form action="{{ login_url }}" method="post">
-        <h2>Se connecter</h2>
-        <input type="hidden" name="came_from" value="{{ came_from }}" />
+    <h2>Se connecter</h2>
 
+    <p>
+        Seuls les comptes Google autorisés par l'administrateur peuvent se connecter au site.<br>
+    </p>
+
+    <form action="{{ login_url }}" method="post">
         <div class="form-group">
-            <input type="text" name="username" class="form-control" placeholder="Identifiant">
-        </div>
-        <div class="form-group">
-            <input type="password" name="password" class="form-control" placeholder="Mot de passe">
-        </div>
-        <div class="form-group">
-            <input type="submit" name="form.submitted" value="Se connecter" class="btn btn-primary">
+            <input type="submit" name="form.submitted" value="Se connecter avec Google" class="btn btn-primary">
         </div>
     </form>
     
+
 	</div>
 </div>
 	<br />

ctp_blogr/templates/user_edit.jinja2

@@ -8,6 +8,10 @@
         </div>
     {% endif %}
 
+    <p>
+        <a class="btn btn-default" href="{{ url_retour }}"><span class="glyphicon glyphicon-chevron-left"></span> Retour</a>
+    </p>
+
     <form action="{{ url }}" method="post" class="form">
 
         {% for error in form.name.errors %}
@@ -16,33 +20,21 @@
 
         <div class="form-group">
             <label class="required-field" for="name">{{form.name.label}}</label>
-            {% if form.id.data  %}
-                <input class="form-control" name="name" readonly type="text" value="{{form.name.data}}">
-            {% else %}
-                {{form.name(class_='form-control')}}
-            {% endif %}
+            <input class="form-control" name="name" readonly type="text" value="{{form.name.data}}">
         </div>
 
-        <div class="form-group">
-            <label class="required-field" for="password">{{form.password.label}}</label>
-            {{form.password(class_='form-control')}}
-        </div>
-
-        {% for error in form.confirm.errors %}
-            <div class="label label-danger">{{error}}</div>
-        {% endfor %}
-
-        <div class="form-group">
-            <label class="required-field" for="confirm">{{form.confirm.label}}</label>
-            {{form.confirm(class_='form-control')}}
-        </div>
+        <p class="help-block">
+            L'adresse {{ form.name.data }} est votre compte Google autorisé. Elle
+            n'est pas modifiable ici : supprimez-la de la base et ajoutez-la à
+            <code>auth.admin_emails</code> ou à la table <code>users</code> pour
+            changer d'identité.
+        </p>
 
         <br>
         <div class="form-group">
-            <a class="btn btn-default" href="{{ url_retour }}"><span class="glyphicon glyphicon-chevron-left"></span> Retour</a>
             <button class="btn btn-primary" type="submit" name="form.submitted">
                 <span class="glyphicon glyphicon-ok"></span> Enregistrer</button>
-            {% if form.id.data and request.authenticated_userid == 'admin' %}
+            {% if 'group:administrators' in request.effective_principals %}
                 <button class="btn btn-warning" type="submit" name="form.deleted">
                     <span class="glyphicon glyphicon-remove"></span> Supprimer</button>
             {% endif %}

ctp_blogr/templates/users.jinja2

@@ -4,8 +4,6 @@
     <p>
         <a href="{{ request.route_url('home' ) }}" class="btn btn-default" role="button">
             <span class="glyphicon glyphicon-chevron-left"></span> Retour</a>
-        <a href="{{ request.route_url('user_edit', name='0') }}" class="btn btn-success" role="button">
-            <span class="glyphicon glyphicon-plus"></span> Nouvel utilisateur</a>
     </p>
 
     <table id="users_list" class="table table-striped table-bordered table-condensed">
@@ -24,7 +22,7 @@
                         {{ entry.name }}
                     </a>
                     </td>
-                <td>{{ entry.last_logged.strftime("%d-%m-%Y - %H:%M") }}</td>
+                <td>{% if entry.last_logged %}{{ entry.last_logged.strftime("%d-%m-%Y - %H:%M") }}{% else %}jamais{% endif %}</td>
             </tr>
         {% endfor %}
     </table>

ctp_blogr/test_oauth.py

@@ -0,0 +1,861 @@
+# Test de bout en bout de la migration Google OAuth, sans aucun acces reseau.
+#
+# Ce module monte une VRAIE application WSGI (Configurator + make_wsgi_app)
+# servie par WebTest, sur une base SQLite en memoire. tests.py ( heritage du
+# scaffold) ne construit qu'un DummyRequest et n'appelle jamais
+# init_database() : il ne couvre donc rien de ce qui suit.
+#
+# Le collaborateur Google est remplace par FakeGoogle : aucun socket n'est
+# ouvert (la fixture automatique 'aucun_acces_reseau' bloque socket.socket),
+# et le nombre d'appels a fetch_claims est compte pour prouver que le point de
+# terminaison de jetons n'est atteint qu'une fois par connexion.
+#
+# Ce module est volontairement long : la consigne impose un seul fichier pour
+# le banc de test et la suite.
+#
+# Les tests DEFAUT_* documentent des defauts connus, anterieurs a la migration
+# OAuth et Decision de ne pas les corriger ici. Ils sont volontairement verts :
+# ils decrivent le comportement observe, pas le comportement souhaite.
+
+import json
+import socket
+import time
+
+from urllib.parse import quote, urlencode
+
+import pytest
+import sqlalchemy as sa
+
+import pyramid.session
+
+from pyramid.authentication import SessionAuthenticationPolicy
+from pyramid.authorization import ACLAuthorizationPolicy
+from pyramid.config import Configurator
+from pyramid.interfaces import ISessionFactory
+from pyramid.session import JSONSerializer, SignedCookieSessionFactory
+from pyramid_retry import RetryableException
+from webob.cookies import SignedSerializer
+from webtest import TestApp
+
+from . import SESSION_SECRET, SESSION_TIMEOUT, admin_emails
+from .models.blog_record import BlogRecord, Tags
+from .models.meta import Base
+from .models.user import User
+from .services.google_oauth import GoogleOAuthError
+from .services.user import make_groupfinder
+
+
+ADMIN_EMAIL = 'phuoc@caotek.fr'
+AUTRE_EMAIL = 'ctphuoc@gmail.com'
+INCONNU_EMAIL = 'inconnu@example.org'
+ADMIN_SUB = '1011222333444555'
+AUTRE_SUB = '5566778899001122'
+INCONNU_SUB = '9998887776665554'
+CLIENT_ID = '123456789.apps.googleusercontent.com'
+CLIENT_SECRET = 'secret-de-test'
+TITRE_BILLET = 'Billets de test'
+SONDE_URL = '/__sonde/session'
+
+
+class HorlogeFausse(object):
+    # Remplace le module time de pyramid.session : permet de vieillir une
+    # session sans time.sleep().
+
+    def __init__(self, valeur):
+        self.valeur = valeur
+
+    def time(self):
+        return self.valeur
+
+
+class ErreurRejouable(RetryableException):
+    # Exception marquee retryable : pyramid_retry rejoue la requete.
+    pass
+
+
+class FakeGoogle(object):
+    # Faux client OIDC : memes deux methodes que celles appellees par
+    # views/default.py. Aucun appel a Google, aucun jeton reellement signe.
+    # fetch_claims reproduit le controle de nonce de
+    # GoogleOAuth.verify_id_token : un nonce renvoye qui ne correspond pas au
+    # nonce de la session fait echouer l'echange.
+
+    def __init__(self, email, sub):
+        self.email = email
+        self.sub = sub
+        # email_verified renvoye par Google (peut valoir False).
+        self.email_verified = True
+        # nonce renvoye par Google : None => le nonce demande est renvoye.
+        self.nonce_recu = None
+        # exception a lever a chaque fetch_claims (None => aucune).
+        self.erreur = None
+        self.fetch_calls = 0
+        self.authorize_calls = 0
+        self.state = None
+        self.nonce = None
+        self.code_verifier = None
+        self.codes = []
+
+    def create_authorization_url(self, redirect_uri, state, nonce, code_verifier):
+        self.authorize_calls += 1
+        self.state = state
+        self.nonce = nonce
+        self.code_verifier = code_verifier
+        params = {
+            'client_id': CLIENT_ID,
+            'response_type': 'code',
+            'redirect_uri': redirect_uri,
+            'scope': 'openid email',
+            'state': state,
+            'nonce': nonce,
+            'code_challenge_method': 'S256',
+        }
+        return 'https://accounts.google.com/o/oauth2/auth?%s' % urlencode(params)
+
+    def fetch_claims(self, code, code_verifier, redirect_uri, nonce):
+        self.fetch_calls += 1
+        self.codes.append(code)
+        if self.erreur is not None:
+            # L'erreur reste posee pour tous les appels suivants : si la
+            # requete etait rejouee, chaque tentative la reverrait.
+            raise self.erreur
+        nonce_recu = nonce if self.nonce_recu is None else self.nonce_recu
+        if nonce_recu != nonce:
+            raise GoogleOAuthError("Le nonce renvoyé par Google ne correspond pas.")
+        return {
+            'sub': self.sub,
+            'email': self.email,
+            'email_verified': self.email_verified,
+        }
+
+
+def sonde_session(request):
+    # Vue de test : rend l'identite, les principalites effectives et le
+    # contenu de la session tels que la requete suivante les voit.
+    session = request.session
+    return {
+        'userid': request.authenticated_userid,
+        'principales': sorted(request.effective_principals),
+        'cles': sorted(session.keys()),
+        'valeurs': dict(session),
+        'tentatives': request.environ.get('retry.attempts'),
+    }
+
+
+def configurer(retry_attempts=1):
+    # Reproduit le câblage de ctp_blogr.main() sans lire os.environ, pour que
+    # le test n depende d'aucune variable d'environnement.
+    settings = {
+        'sqlalchemy.url': 'sqlite:///:memory:',
+        'google.client_id': CLIENT_ID,
+        'google.client_secret': CLIENT_SECRET,
+        'auth.admin_emails': ADMIN_EMAIL,
+        'session.cookie_secure': 'false',
+        'retry.attempts': str(retry_attempts),
+    }
+    factory = SignedCookieSessionFactory(
+        SESSION_SECRET,
+        timeout=SESSION_TIMEOUT,
+        httponly=True,
+        secure=False,
+    )
+    with Configurator(
+            settings=settings,
+            root_factory='ctp_blogr.security.RootFactory',
+            authentication_policy=SessionAuthenticationPolicy(
+                callback=make_groupfinder(admin_emails(settings))),
+            authorization_policy=ACLAuthorizationPolicy()) as config:
+        config.include('pyramid_jinja2')
+        config.include('.models')
+        config.include('.routes')
+        config.include('.services.google_oauth')
+        config.set_session_factory(factory)
+        config.add_route('sonde_session', SONDE_URL)
+        config.add_view(sonde_session, route_name='sonde_session', renderer='json')
+        config.scan('ctp_blogr.views')
+    return config.make_wsgi_app(), factory
+
+
+def semer(engine):
+    # Deux utilisateurs autorises, aucun sub Google, aucune derniere connexion,
+    # plus un billet publie pour les tests d'edition.
+    with engine.begin() as conn:
+        conn.execute(sa.insert(User), [
+            {'id': 1, 'name': ADMIN_EMAIL, 'google_sub': None, 'last_logged': None},
+            {'id': 2, 'name': AUTRE_EMAIL, 'google_sub': None, 'last_logged': None},
+        ])
+        conn.execute(sa.insert(Tags), [{'id': 1, 'tag': 'test'}])
+        conn.execute(sa.insert(BlogRecord), [{
+            'id': 1,
+            'title': TITRE_BILLET,
+            'body': 'Un billet de test.',
+            'creator': ADMIN_EMAIL,
+            'editor': ADMIN_EMAIL,
+            'tag': 'test',
+            'status': 'publié',
+        }])
+
+
+class Blog(object):
+    # Application de test : une WSGI app, un moteur, le faux client Google et
+    # quelques lectures de base pour verifier le contenu et non le statut seul.
+
+    def __init__(self, wsgi_app, engine, google, factory):
+        self.wsgi_app = wsgi_app
+        self.engine = engine
+        self.google = google
+        self.factory = factory
+
+    def client(self):
+        # Chaque client a son propre cookie jar : un administrateur, un
+        # utilisateur simple et un visiteur anonyme sont trois clients.
+        return TestApp(self.wsgi_app)
+
+    def rows(self):
+        with self.engine.connect() as conn:
+            return [dict(row._mapping) for row in conn.execute(sa.select(User))]
+
+    def utilisateur(self, nom):
+        for ligne in self.rows():
+            if ligne['name'] == nom:
+                return ligne
+        return None
+
+    def nb_users(self):
+        with self.engine.connect() as conn:
+            return conn.execute(
+                sa.text('SELECT COUNT(*) FROM users')).scalar()
+
+    def billet(self, billet_id=1):
+        with self.engine.connect() as conn:
+            ligne = conn.execute(
+                sa.text('SELECT * FROM entries WHERE id = :id'),
+                {'id': billet_id}).mappings().first()
+        return dict(ligne) if ligne is not None else None
+
+    def etat_session(self, valeur_cookie):
+        # Le cookie signe contient (renouvele, cree, etat) : on le decode avec
+        # le meme secret que l'application.
+        signer = SignedSerializer(SESSION_SECRET, 'pyramid.session.', 'sha512',
+                                  serializer=JSONSerializer())
+        return signer.loads(valeur_cookie.encode('utf-8'))[2]
+
+    def factory_session(self):
+        return self.wsgi_app.registry.queryUtility(ISessionFactory)
+
+
+def demarrer(retry_attempts=1):
+    wsgi_app, factory = configurer(retry_attempts=retry_attempts)
+    engine = wsgi_app.registry['dbsession_factory'].kw['bind']
+    google = FakeGoogle(email=ADMIN_EMAIL, sub=ADMIN_SUB)
+    # includeme ecrit l'ATTRIBUT, pas une cle de dict. Les deux sont poses ici
+    # pour que le banc de test ne depende pas de cette distinction.
+    wsgi_app.registry['google_oauth'] = google
+    wsgi_app.registry.google_oauth = google
+    Base.metadata.create_all(engine)
+    semer(engine)
+    return Blog(wsgi_app, engine, google, factory)
+
+
+@pytest.fixture(autouse=True)
+def aucun_acces_reseau(monkeypatch):
+    # Preuve d'absence de reseau : aucune socket ne peut etre ouverte pendant
+    # ce module. SQLite (:memory:) et WebTest n'en ont pas besoin.
+    def refuser(*args, **kwargs):
+        raise AssertionError('acces reseau interdit dans test_oauth.py')
+
+    monkeypatch.setattr(socket, 'socket', refuser)
+    monkeypatch.setattr(socket, 'create_connection', refuser)
+    monkeypatch.setattr(socket, 'getaddrinfo', refuser)
+
+
+@pytest.fixture
+def blog():
+    return demarrer()
+
+
+def url_callback(blog, code='code-de-test', state=None):
+    if state is None:
+        state = blog.google.state
+    return '/login_google_callback?code=%s&state=%s' % (code, quote(state))
+
+
+def cookie_session(client):
+    for cookie in client.cookiejar:
+        if cookie.name == 'session':
+            return cookie.value
+    return None
+
+
+def demarrer_connexion(blog, client):
+    # Premier saut : envoi du formulaire de connexion (POST /login avec
+    # form.submitted), redirection vers Google et mise en session de
+    # state / nonce / code_verifier.
+    res = client.post('/login',
+                      {'form.submitted': 'Se connecter avec Google'},
+                      status=302)
+    assert 'accounts.google.com' in res.location
+    assert blog.google.authorize_calls == 1
+    assert blog.google.nonce and blog.google.code_verifier
+    return res
+
+
+def connecter(blog, email=ADMIN_EMAIL, sub=ADMIN_SUB):
+    client = blog.client()
+    blog.google.email = email
+    blog.google.sub = sub
+    demarrer_connexion(blog, client)
+    client.get(url_callback(blog), status=302)
+    return client
+
+
+def lire_sonde(client):
+    return client.get(SONDE_URL, status=200).json
+
+
+# 1. page de connexion
+
+
+def test_page_de_connexion_ne_propose_qu_un_bouton_google(blog):
+    res = blog.client().get('/login')
+    assert res.status_int == 200
+    assert 'Se connecter avec Google' in res.text
+    # Un seul formulaire : un POST vers /login qui porte form.submitted,
+    # pas de lien vers une route de depart dediee (elle n'existe pas).
+    assert '<form action="http://localhost/login" method="post">' in res.text
+    assert 'name="form.submitted"' in res.text
+    assert '<input type="password"' not in res.text
+    assert 'name="password"' not in res.text
+
+
+# 2 et 3. les deux parcours heureux
+
+
+def test_connexion_administrateur_renseigne_le_sub_et_la_derniere_connexion(blog):
+    avant = blog.utilisateur(ADMIN_EMAIL)
+    assert avant['last_logged'] is None
+    assert avant['google_sub'] is None
+
+    client = blog.client()
+    depart = demarrer_connexion(blog, client)
+    assert 'state=%s' % quote(blog.google.state) in depart.location
+
+    res = client.get(url_callback(blog), status=302)
+    assert res.location == 'http://localhost/'
+    assert blog.google.fetch_calls == 1
+    assert 'users_list' not in res.text
+
+    etat = blog.etat_session(cookie_session(client))
+    assert etat['auth.userid'] == ADMIN_EMAIL
+    assert etat['_f_success'] == ['Bienvenue %s !' % ADMIN_EMAIL]
+
+    sonde = lire_sonde(client)
+    assert sonde['userid'] == ADMIN_EMAIL
+    assert 'group:administrators' in sonde['principales']
+
+    apres = blog.utilisateur(ADMIN_EMAIL)
+    assert apres['google_sub'] == ADMIN_SUB
+    assert apres['last_logged'] is not None
+    assert apres['google_sub'] != avant['google_sub']
+
+
+def test_connexion_non_administrateur_sans_droit_sur_la_liste_des_utilisateurs(blog):
+    client = connecter(blog, AUTRE_EMAIL, AUTRE_SUB)
+
+    sonde = lire_sonde(client)
+    assert sonde['userid'] == AUTRE_EMAIL
+    assert 'group:administrators' not in sonde['principales']
+    assert 'system.Authenticated' in sonde['principales']
+    assert 'system.Everyone' in sonde['principales']
+
+    ligne = blog.utilisateur(AUTRE_EMAIL)
+    assert ligne['google_sub'] == AUTRE_SUB
+    assert ligne['last_logged'] is not None
+
+    res = client.get('/users', status=403)
+    assert res.status_int == 403
+    assert 'users_list' not in res.text
+    assert 'Accès refusé' in res.text
+
+
+# 4. liste blanche : aucun compte cree automatiquement
+
+
+def test_compte_google_inconnu_refuse_sans_creation_de_fiche(blog):
+    avant = blog.nb_users()
+    assert avant == 2
+
+    client = blog.client()
+    blog.google.email = INCONNU_EMAIL
+    blog.google.sub = INCONNU_SUB
+    demarrer_connexion(blog, client)
+
+    res = client.get(url_callback(blog), status=403)
+    assert res.status_int == 403
+    assert 'est pas autorisé sur ce blog' in res.text
+    assert 'Demandez son ajout' in res.text
+    assert INCONNU_EMAIL in res.text
+    assert '<input type="password"' not in res.text
+    assert 'name="form.submitted"' not in res.text
+
+    sonde = lire_sonde(client)
+    assert sonde['userid'] is None
+    assert blog.nb_users() == avant
+    for ligne in blog.rows():
+        assert ligne['google_sub'] is None
+        assert ligne['last_logged'] is None
+
+
+# 5. state : absent, incorrect, rejoue
+
+
+def test_state_absent_refuse_la_connexion_sans_echange(blog):
+    client = blog.client()
+    res = client.get('/login_google_callback?code=code-de-test', status=302)
+    assert res.location == 'http://localhost/login'
+    assert blog.google.fetch_calls == 0
+
+    page = client.get('/login', status=200)
+    assert 'Requête de connexion invalide ou déjà utilisée.' in page.text
+    assert lire_sonde(client)['userid'] is None
+
+
+def test_state_incorrect_refuse_la_connexion_sans_echange(blog):
+    client = blog.client()
+    demarrer_connexion(blog, client)
+
+    res = client.get(url_callback(blog, state='state-fabrique-par-un-tiers'), status=302)
+    assert res.location == 'http://localhost/login'
+    assert blog.google.fetch_calls == 0
+
+    page = client.get('/login', status=200)
+    assert 'Requête de connexion invalide ou déjà utilisée.' in page.text
+    assert lire_sonde(client)['userid'] is None
+    assert blog.utilisateur(ADMIN_EMAIL)['last_logged'] is None
+
+
+def test_state_rejoue_refuse_la_connexion_sans_echange(blog):
+    client = blog.client()
+    demarrer_connexion(blog, client)
+
+    # Premier passage : le client Google refuse, l'etat est consomme.
+    refuse = client.get(
+        '/login_google_callback?error=access_denied&state=%s'
+        % quote(blog.google.state), status=302)
+    assert refuse.location == 'http://localhost/login'
+    assert blog.google.fetch_calls == 0
+
+    # Rejeu du meme state avec un code : plus rien en session, donc refus.
+    rejeu = client.get(url_callback(blog), status=302)
+    assert rejeu.location == 'http://localhost/login'
+    assert blog.google.fetch_calls == 0
+
+    page = client.get('/login', status=200)
+    assert 'Requête de connexion invalide ou déjà utilisée.' in page.text
+    assert lire_sonde(client)['userid'] is None
+
+
+def test_state_rejoue_apres_connexion_reussie_n_echange_pas_deux_fois(blog):
+    client = connecter(blog, ADMIN_EMAIL, ADMIN_SUB)
+    assert blog.google.fetch_calls == 1
+
+    rejeu = client.get(url_callback(blog), status=302)
+    assert rejeu.location == 'http://localhost/login'
+    assert blog.google.fetch_calls == 1
+    assert blog.google.codes == ['code-de-test']
+    assert lire_sonde(client)['userid'] == ADMIN_EMAIL
+
+
+# 6. nonce
+
+
+def test_nonce_non_concordant_refuse_la_connexion(blog):
+    client = blog.client()
+    demarrer_connexion(blog, client)
+    blog.google.nonce_recu = 'nonce-fabrique-par-un-tiers'
+
+    res = client.get(url_callback(blog), status=302)
+    assert res.location == 'http://localhost/login'
+    assert blog.google.fetch_calls == 1
+
+    page = client.get('/login', status=200)
+    assert 'Connexion Google échouée' in page.text
+    assert 'nonce' in page.text
+    assert lire_sonde(client)['userid'] is None
+    assert blog.utilisateur(ADMIN_EMAIL)['google_sub'] is None
+
+
+# 6 bis. claims de l'ID token : code, email_verified, sub
+
+
+def test_code_absent_refuse_la_connexion_sans_echange(blog):
+    client = blog.client()
+    demarrer_connexion(blog, client)
+
+    res = client.get(
+        '/login_google_callback?state=%s' % quote(blog.google.state), status=302)
+    assert res.location == 'http://localhost/login'
+    assert blog.google.fetch_calls == 0
+
+    page = client.get('/login', status=200)
+    assert 'Requête de connexion incomplète.' in page.text
+    assert lire_sonde(client)['userid'] is None
+
+
+def test_adresse_non_verifiee_refusee_meme_si_elle_est_autorisee(blog):
+    # L'identite EST l'adresse : accepter une adresse Google non verifiee
+    # permettrait d'usurper un compte de la liste blanche.
+    client = blog.client()
+    blog.google.email = ADMIN_EMAIL
+    blog.google.sub = ADMIN_SUB
+    blog.google.email_verified = False
+    demarrer_connexion(blog, client)
+
+    res = client.get(url_callback(blog), status=302)
+    assert res.location == 'http://localhost/login'
+    assert blog.google.fetch_calls == 1
+
+    page = client.get('/login', status=200)
+    assert 'pas vérifiée' in page.text
+    assert lire_sonde(client)['userid'] is None
+    ligne = blog.utilisateur(ADMIN_EMAIL)
+    assert ligne['google_sub'] is None
+    assert ligne['last_logged'] is None
+
+
+def test_reponse_sans_sub_refusee_et_ne_detruit_pas_le_google_sub(blog):
+    # Sans ce refus, la ligne authorizee recevait google_sub='' : la contrainte
+    # UNIQUE interdisait alors le second compte sans sub, et la fiche devenait
+    # rebindable par tout autre compte portant la meme adresse.
+    client = blog.client()
+    blog.google.email = ADMIN_EMAIL
+    blog.google.sub = ''
+    demarrer_connexion(blog, client)
+
+    res = client.get(url_callback(blog), status=302)
+    assert res.location == 'http://localhost/login'
+    assert blog.google.fetch_calls == 1
+
+    page = client.get('/login', status=200)
+    assert 'Réponse Google incomplète.' in page.text
+    assert lire_sonde(client)['userid'] is None
+    assert blog.utilisateur(ADMIN_EMAIL)['google_sub'] is None
+
+
+# 7. refus de l'utilisateur chez Google
+
+
+def test_refus_de_l_utilisateur_chez_google_affiche_un_message(blog):
+    client = blog.client()
+    demarrer_connexion(blog, client)
+
+    res = client.get(
+        '/login_google_callback?error=access_denied&state=%s'
+        % quote(blog.google.state), status=302)
+    assert res.location == 'http://localhost/login'
+    assert blog.google.fetch_calls == 0
+
+    page = client.get('/login', status=200)
+    assert 'Connexion Google annulée ou refusée.' in page.text
+    assert 'Se connecter avec Google' in page.text
+    assert '<input type="password"' not in page.text
+
+    sonde = lire_sonde(client)
+    assert sonde['userid'] is None
+    assert blog.nb_users() == 2
+    for ligne in blog.rows():
+        assert ligne['google_sub'] is None
+        assert ligne['last_logged'] is None
+
+
+# 8. securite du rejeu (pyramid_retry)
+
+
+def test_le_rejeu_de_la_requete_n_appelle_qu_une_fois_le_point_de_jetons():
+    blog = demarrer(retry_attempts=3)
+    assert blog.wsgi_app.registry.settings['retry.attempts'] == 3
+
+    client = blog.client()
+    demarrer_connexion(blog, client)
+    # Erreur transitoire telle que GoogleOAuth la leve : echange de jetons
+    # refuse, la vue transforme l'echec en redirection.
+    blog.google.erreur = GoogleOAuthError(
+        "Google a refusé le code d'autorisation : <urlopen error timed out>")
+
+    res = client.get(url_callback(blog), status=302)
+    assert res.location == 'http://localhost/login'
+    assert blog.google.fetch_calls == 1
+    assert blog.google.codes == ['code-de-test']
+
+    page = client.get('/login', status=200)
+    assert 'Connexion Google échouée' in page.text
+    sonde = lire_sonde(client)
+    assert sonde['userid'] is None
+    assert sonde['tentatives'] == 3
+
+
+# 9. la session est nettoyee
+
+
+def test_la_session_ne_conserve_plus_state_nonce_code_verifier(blog):
+    client = blog.client()
+    demarrer_connexion(blog, client)
+    secrets = [blog.google.state, blog.google.nonce, blog.google.code_verifier]
+    client.get(url_callback(blog), status=302)
+
+    sonde = lire_sonde(client)
+    for cle in sonde['cles']:
+        for mot in ('state', 'nonce', 'code_verifier'):
+            assert mot not in cle
+    for secret in secrets:
+        assert secret not in json.dumps(sonde['valeurs'])
+        assert secret not in json.dumps(blog.etat_session(cookie_session(client)))
+    assert sonde['cles'] == ['_f_success', 'auth.userid']
+
+
+# 10. un utilisateur simple n'ouvre pas la fiche d'un autre
+
+
+def test_non_administrateur_ne_peut_pas_ouvrir_la_fiche_d_un_autre(blog):
+    client = connecter(blog, AUTRE_EMAIL, AUTRE_SUB)
+
+    res = client.get('/user_edit/%s' % ADMIN_EMAIL, status=403)
+    assert res.status_int == 403
+    assert 'Accès refusé' in res.text
+    assert 'Modification utilisateur' not in res.text
+    assert 'name="form.submitted"' not in res.text
+    assert 'name="name"' not in res.text
+    assert '<form action=' not in res.text
+
+
+def test_non_administrateur_ne_peut_pas_creer_de_fiche(blog):
+    client = connecter(blog, AUTRE_EMAIL, AUTRE_SUB)
+
+    res = client.get('/user_edit/0', status=403)
+    assert res.status_int == 403
+    assert 'Accès refusé' in res.text
+    assert 'Utilisateur non trouvé' not in res.text
+    assert 'name="form.submitted"' not in res.text
+    assert blog.nb_users() == 2
+
+
+# 11. fiche administrateur : affichage et suppression
+
+
+def test_administrateur_voit_la_fiche_sans_champ_mot_de_passe(blog):
+    client = connecter(blog, ADMIN_EMAIL, ADMIN_SUB)
+
+    res = client.get('/user_edit/%s' % ADMIN_EMAIL)
+    assert res.status_int == 200
+    assert 'Modification utilisateur' in res.text
+    assert ADMIN_EMAIL in res.text
+    assert 'name="password"' not in res.text
+    assert 'name="form.submitted"' in res.text
+    assert 'name="form.deleted"' in res.text
+
+
+def test_administrateur_supprime_une_fiche_sans_erreur(blog):
+    client = connecter(blog, ADMIN_EMAIL, ADMIN_SUB)
+    assert blog.nb_users() == 2
+
+    res = client.post('/user_edit/%s' % ADMIN_EMAIL,
+                      {'form.deleted': 'Supprimer'}, status=302)
+    assert res.location == 'http://localhost/users'
+    assert blog.nb_users() == 1
+    assert blog.utilisateur(ADMIN_EMAIL) is None
+    assert blog.utilisateur(AUTRE_EMAIL) is not None
+
+    page = client.get('/users', status=200)
+    assert 'La fiche a été supprimée avec succès.' in page.text
+    # La barre de navigation affiche encore l'identite porte par la session :
+    # c'est le tableau users_list qui doit avoir perdu la ligne supprimee.
+    assert '/user_edit/%s' % ADMIN_EMAIL not in page.text
+    assert '/user_edit/%s' % AUTRE_EMAIL in page.text
+
+
+# 12. liste des utilisateurs et derniere connexion vide
+
+
+def test_liste_des_utilisateurs_gere_une_derniere_connexion_vide(blog):
+    client = connecter(blog, ADMIN_EMAIL, ADMIN_SUB)
+    vide = [ligne for ligne in blog.rows() if ligne['last_logged'] is None]
+    assert [ligne['name'] for ligne in vide] == [AUTRE_EMAIL]
+
+    res = client.get('/users')
+    assert res.status_int == 200
+    assert 'users_list' in res.text
+    assert 'jamais' in res.text
+    assert '>None<' not in res.text
+    assert AUTRE_EMAIL in res.text
+    assert ADMIN_EMAIL in res.text
+
+
+def test_page_des_utilisateurs_refusee_au_visiteur_non_connecte(blog):
+    res = blog.client().get('/users', status=403)
+    assert res.status_int == 403
+    assert 'users_list' not in res.text
+    assert 'Accès refusé' in res.text
+
+
+# 13. edition d'un billet
+
+
+def test_visiteur_non_connecte_ne_peut_pas_modifier_un_billet(blog):
+    client = blog.client()
+
+    res = client.post('/blog_edit/1', {
+        'title': 'Titre vole',
+        'body': 'corps vole',
+        'tag': 'test',
+        'status': 'publié',
+        'form.submitted': 'Enregistrer',
+    }, status=403)
+    assert res.status_int == 403
+    assert 'Accès refusé' in res.text
+    assert 'name="form.submitted"' not in res.text
+
+    billet = blog.billet(1)
+    assert billet['title'] == TITRE_BILLET
+    assert billet['editor'] == ADMIN_EMAIL
+    page = client.get('/blog/1/billets-de-test', status=200)
+    assert TITRE_BILLET in page.text
+
+
+def test_visiteur_non_connecte_ne_peut_pas_supprimer_un_billet(blog):
+    client = blog.client()
+
+    res = client.post('/blog_edit/1', {'form.deleted': 'Supprimer'}, status=403)
+    assert res.status_int == 403
+    assert 'name="form.deleted"' not in res.text
+
+    billet = blog.billet(1)
+    assert billet is not None
+    assert billet['title'] == TITRE_BILLET
+    page = client.get('/blog/1/billets-de-test', status=200)
+    assert TITRE_BILLET in page.text
+
+
+# 14. duree et Attributes de la session
+
+
+def test_timeout_de_session_laisse_plus_de_1200_secondes(blog):
+    factory = blog.factory_session()
+    assert factory is blog.factory
+    assert factory._timeout == SESSION_TIMEOUT
+    assert factory._timeout > 1200
+    assert factory._cookie_httponly is True
+
+
+def test_session_vieille_de_1199_secondes_encore_valide(blog, monkeypatch):
+    client = connecter(blog, ADMIN_EMAIL, ADMIN_SUB)
+    assert lire_sonde(client)['userid'] == ADMIN_EMAIL
+
+    monkeypatch.setattr(
+        pyramid.session, 'time', HorlogeFausse(time.time() + 1199))
+    assert lire_sonde(client)['userid'] == ADMIN_EMAIL
+    assert 'users_list' in client.get('/users', status=200).text
+
+
+def test_session_expiree_est_refusee_avec_une_horloge_simulee(blog, monkeypatch):
+    client = connecter(blog, ADMIN_EMAIL, ADMIN_SUB)
+    assert lire_sonde(client)['userid'] == ADMIN_EMAIL
+
+    monkeypatch.setattr(
+        pyramid.session, 'time', HorlogeFausse(time.time() + SESSION_TIMEOUT + 1))
+
+    sonde = lire_sonde(client)
+    assert sonde['userid'] is None
+    assert sonde['cles'] == []
+    assert 'group:administrators' not in sonde['principales']
+
+    res = client.get('/users', status=403)
+    assert res.status_int == 403
+    assert 'Accès refusé' in res.text
+    assert 'users_list' not in res.text
+
+
+def test_cookie_de_session_httponly(blog):
+    client = blog.client()
+    res = client.post('/login',
+                      {'form.submitted': 'Se connecter avec Google'},
+                      status=302)
+    entetes = res.headers.getall('Set-Cookie')
+    assert entetes
+    assert any('session=' in entete and 'HttpOnly' in entete for entete in entetes)
+    assert blog.factory_session()._cookie_httponly is True
+
+
+# 15. le garde-fou reseau est bien actif
+
+
+def test_le_garde_fou_reseau_bloque_les_sockets():
+    with pytest.raises(AssertionError):
+        socket.socket()
+    with pytest.raises(AssertionError):
+        socket.create_connection(('accounts.google.com', 443))
+    with pytest.raises(AssertionError):
+        socket.getaddrinfo('accounts.google.com', 443)
+
+
+# ANOMALIES de la migration, decrites telles qu'elles se comportent
+
+
+def test_le_client_google_est_pose_comme_attribut_du_registry():
+    # services/google_oauth.py::includeme ecrit l'attribut, pas la cle :
+    # request.registry.google_oauth leveait AttributeError sinon, et la page
+    # /login se rendait bien en GET alors que son envoi (POST, qui lance le
+    # flux vers Google) plantait.
+    wsgi_app, _ = configurer()
+    assert type(wsgi_app.registry.google_oauth).__name__ == 'GoogleOAuth'
+
+
+def test_la_redirection_vers_google_est_atteignable():
+    blog = demarrer()
+    client = blog.client()
+    res = client.post('/login',
+                      {'form.submitted': 'Se connecter avec Google'},
+                      status=302)
+    assert res.status_int == 302
+    assert res.location.startswith('https://accounts.google.com/')
+    assert 'code_challenge_method=S256' in res.location
+    assert 'scope=openid+email' in res.location
+    assert 'nonce=' in res.location
+    assert 'hd=' not in res.location
+
+
+def test_une_erreur_rejouable_ne_rejoue_pas_le_point_de_jetons():
+    # La vue se declare derniere tentative avant l'appel reseau : sinon
+    # pyramid_retry relit le state dans le cookie d'origine et renvoie le meme
+    # code d'autorisation une fois par tentative (invalid_grant).
+    blog = demarrer(retry_attempts=3)
+    client = blog.client()
+    demarrer_connexion(blog, client)
+    blog.google.erreur = ErreurRejouable('panne reseau transitoire')
+
+    with pytest.raises(ErreurRejouable):
+        client.get(url_callback(blog), status='*')
+
+    assert blog.google.fetch_calls == 1
+    assert lire_sonde(client)['userid'] is None
+
+
+def test_DEFAUT_connu_un_utilisateur_non_administrateur_peut_modifier_un_billet(blog):
+    # security.py accorde 'view' a tout utilisateur connecte : un compte
+    # autorise mais non administrateur passe dans blog_edit, qui n'exige que
+    # 'view'. Seul l'anonyme est refuse (cf. les deux tests precedents).
+    # Decision de ne pas corriger ici : le README prevoit un role de
+    # redacteur, et blog_edit n'exige que 'view' depuis toujours.
+    client = connecter(blog, AUTRE_EMAIL, AUTRE_SUB)
+
+    res = client.post('/blog_edit/1', {
+        'title': 'Titre modifie',
+        'body': 'corps modifie',
+        'tag': 'test',
+        'status': 'publié',
+        'form.submitted': 'Enregistrer',
+    }, status=302)
+    assert res.location == 'http://localhost/blog/1/titre-modifie'
+
+    billet = blog.billet(1)
+    assert billet['title'] == 'Titre modifie'
+    assert billet['editor'] == AUTRE_EMAIL

ctp_blogr/views/default.py

@@ -1,13 +1,13 @@
-from pyramid.view import (
-    view_config,
-    forbidden_view_config,
-)
-from pyramid.httpexceptions import HTTPFound
+import datetime
+
+from authlib.common.security import generate_token
+from pyramid.view import view_config
+from pyramid.httpexceptions import HTTPForbidden, HTTPFound
 from pyramid.security import remember, forget
+from ..services.google_oauth import GoogleOAuthError
 from ..services.user import UserService
 from ..services.blog_record import BlogRecordService
-from ..forms import UserCreateForm
-from ..models.user import User
+from ..forms import UserEditForm
 
 
 @view_config(route_name='home', renderer='../templates/home.jinja2')
@@ -35,36 +35,138 @@ def apropos(request):
 
 
 @view_config(route_name='login', renderer='../templates/login.jinja2')
-@forbidden_view_config(renderer='../templates/login.jinja2')
 def login(request):
     login_url = request.route_url('login')
     
-    referrer = request.url
-    if referrer == login_url:
-        referrer = '/' # never use the login form itself as came_from
-    
-    came_from = request.params.get('came_from', referrer)
-    username = ''
-    userpwd = ''
     if 'form.submitted' in request.params:
-        username = request.POST.get('username')
-        userpwd = request.POST.get('password')
-        user = UserService.by_name(request, username)
-        if user and user.verify_password(userpwd):
-            headers = remember(request, username)
-            request.session.flash("Bienvenue %s !" % username, 'success')
-            return HTTPFound(location=came_from, headers=headers)
-        else:
-            headers = forget(request)
-            request.session.flash("Login et mot de passe invalides. La connexion a échoué.", "danger")
+        session = request.session
+        state = generate_token(32)
+        nonce = generate_token(32)
+        code_verifier = generate_token(48)
+
+        session['oauth_state'] = state
+        session['oauth_nonce'] = nonce
+        session['oauth_code_verifier'] = code_verifier
+
+        url = request.registry.google_oauth.create_authorization_url(
+            redirect_uri=request.route_url('login_google_callback'),
+            state=state,
+            nonce=nonce,
+            code_verifier=code_verifier,
+        )
+        return HTTPFound(location=url)
      
     return {
         'page_title': "",
-        'came_from': came_from,
         'login_url': login_url,
         }
 
 
+@view_config(context=HTTPForbidden,
+             renderer='../templates/denied.jinja2')
+def denied(request):
+    # Vrai 403. Un forbidden_view_config sur la page de connexion rendait le
+    # statut 200 et laissait croire que la connexion locale etait possible.
+    request.response.status_int = 403
+    return {
+        'page_title': "Accès refusé",
+        'message': getattr(request, 'denied_message', None)
+                    or "Vous n'avez pas les droits nécessaires pour cette page.",
+        }
+
+@view_config(route_name='login_google_callback',
+             renderer='../templates/denied.jinja2')
+def login_google_callback(request):
+    session = request.session
+
+    def forget_oauth_values():
+        session.pop('oauth_state', None)
+        session.pop('oauth_nonce', None)
+        session.pop('oauth_code_verifier', None)
+
+    def back_to_login(message):
+        forget_oauth_values()
+        request.session.flash(message, 'danger')
+        return HTTPFound(location=request.route_url('login'))
+
+    if request.params.get('error'):
+        return back_to_login("Connexion Google annulée ou refusée.")
+
+    # state, nonce et code_verifier sont consommes AVANT l'echange du code, donc
+    # un rejeu de la requete ne trouve plus de state et n'atteint pas Google.
+    state = session.pop('oauth_state', None)
+    nonce = session.pop('oauth_nonce', None)
+    code_verifier = session.pop('oauth_code_verifier', None)
+
+    if not state or state != request.params.get('state'):
+        return back_to_login("Requête de connexion invalide ou déjà utilisée.")
+
+    if not code_verifier or not nonce:
+        return back_to_login("Requête de connexion incomplète.")
+
+    code = request.params.get('code')
+    if not code:
+        return back_to_login("Requête de connexion incomplète.")
+
+    # A partir d'ici le code d'autorisation peut avoir ete consomme par Google,
+    # alors qu'un rejeu repart du COOKIE D'ORIGINE, qui porte encore oauth_state
+    # et renvoierait le meme code une deuxieme fois, en invalid_grant. On se
+    # declare donc derniere tentative avant le moindre appel reseau.
+    attempt = request.environ.get('retry.attempt')
+    if attempt is not None:
+        request.environ['retry.attempts'] = attempt + 1
+
+    try:
+        claims = request.registry.google_oauth.fetch_claims(
+            code=code,
+            code_verifier=code_verifier,
+            redirect_uri=request.route_url('login_google_callback'),
+            nonce=nonce,
+        )
+    except GoogleOAuthError as exc:
+        return back_to_login("Connexion Google échouée : %s" % exc)
+
+    email = (claims.get('email') or '').strip()
+    google_sub = (claims.get('sub') or '').strip()
+
+    # `email_verified` vaut False pour un compte Google dont l'adresse n'a pas
+    # ete verifiee. Sans ce refus, un compte non verifie pourrait se presenter
+    # avec une adresse de la liste blanche : l'identite EST l'adresse, donc
+    # accepter une adresse non verifiee equivaut a usurper un compte autorise.
+    if claims.get('email_verified') is not True:
+        return back_to_login("L'adresse Google renvoyée n'est pas vérifiée.")
+
+    # Sans `sub` on ne peut pas relier la session a un compte Google precis :
+    # la ligne authorizee se retrouverait avec un google_sub vide, donc
+    # rebindable par n'importe quel autre compte portant la meme adresse.
+    if not google_sub:
+        return back_to_login("Réponse Google incomplète.")
+
+    # Liste blanche : la table users fait foi. Aucune ligne n'est creee pour un
+    # compte inconnu.
+    user = UserService.by_google_sub(request, google_sub)
+
+    if user is None and email:
+        user = UserService.by_name(request, email)
+    if user is None:
+        # 403 explicite, sans lever HTTPForbidden, sinon on retombe sur le
+        # gabarit de connexion et l'appelant ne voit qu'un 200.
+        request.response.status_int = 403
+        return {
+            'page_title': "Accès refusé",
+            'message': ("Le compte Google %s n'est pas autorisé sur ce blog. "
+                        "Demandez son ajout à un administrateur." % (email or google_sub)),
+            }
+
+    if user.google_sub != google_sub:
+        user.google_sub = google_sub
+    user.last_logged = datetime.datetime.now()
+
+    headers = remember(request, user.name)
+    request.session.flash("Bienvenue %s !" % user.name, 'success')
+    return HTTPFound(location=request.route_url('home'), headers=headers)
+
+
 @view_config(route_name='logout', renderer='string')
 def logout(request):
     username = request.authenticated_userid
@@ -85,61 +187,40 @@ def users(request):
         'users': users
         }
 
-@view_config(route_name='user_edit', renderer='ctp_blogr:templates/user_edit.jinja2', permission='view')
+
+@view_config(route_name='user_edit',
+             renderer='ctp_blogr:templates/user_edit.jinja2',
+             permission='manage')
 def user_edit(request):
+    # 'manage' et non 'view' : avec 'view', n'importe quel utilisateur
+    # connecte pouvait ouvrir /user_edit/<un-autre-email> et le modifier,
+    # ou /user_edit/0 pour creer une fiche.
     message = ''
     name = request.matchdict['name']
     url = request.route_url('user_edit', name=name)
-    if request.authenticated_userid == 'admin':
+    if 'group:administrators' in request.effective_principals:
         url_retour = request.route_url('users')
     else:
         url_retour = request.route_url('home')
 
-    if name == '0':
-        # nouvel utilisateur
-        user = User()
-        form = UserCreateForm(request.POST, user)
-        page_title = "Nouvel utilisateur"
-    else:
-        # lire la fiche du user 
-        user = UserService.by_name(request, name)
-        if not user:
-            request.session.flash("Utilisateur non trouvé : %s" % name, 'danger')
-            return HTTPFound(location=url_retour)
-
-        form = UserCreateForm(request.POST, user)
-        page_title = "Modification utilisateur"
+    user = UserService.by_name(request, name)
+    if not user:
+        # '0' n'existe plus : la création d'un utilisateur a disparu avec
+        # le mot de passe.
+        request.session.flash("Utilisateur non trouvé : %s" % name, 'danger')
+        return HTTPFound(location=url_retour)
 
+    form = UserEditForm(request.POST, user)
+    page_title = "Modification utilisateur"
 
     if 'form.submitted' in request.params and form.validate():
-        # controle que le password a moins 6 car
-        if len(form.password.data) < 6 :
-            message = "Le mot de passe doit avoir au moins 6 caractères"
-        else:
-            if name == '0':
-                # création user                
-                # controler que le nouvel user n'existe pas dans la BD 
-                new_user = UserService.by_name(request, form.name.data)
-                if new_user:
-                    message = "Utilisateur déjà créé : %s" % form.name.data
-                else:
-                    form.populate_obj(user)
-                    user.set_password(form.password.data.encode('utf8'))
-                    # créer le nouveau
-                    request.dbsession.add(user)
-                    request.session.flash("La fiche a été créée avec succès.", 'success')
-                    return HTTPFound(location=url_retour)
-
-            else:
-                # modification user
-                del form.name  # SECURITY: prevent overwriting of primary key
-                form.populate_obj(user)
-                user.set_password(form.password.data.encode('utf8'))
-                request.session.flash("La fiche a été modifiée avec succès.", 'success')
-                return HTTPFound(location=url_retour)
+        del form.name  # SECURITY: prevent overwriting of primary key
+        form.populate_obj(user)
+        request.session.flash("La fiche a été modifiée avec succès.", 'success')
+        return HTTPFound(location=url_retour)
 
     if 'form.deleted' in request.params:
-        delete(request, user.id)
+        UserService.delete(request, user.id)
         request.session.flash("La fiche a été supprimée avec succès.", 'success')
         return HTTPFound(location=url_retour)
 

development.ini

@@ -18,6 +18,18 @@ sqlalchemy.url = sqlite:///%(here)s/ctp_blogr.sqlite
 
 retry.attempts = 3
 
+# Les identifiants OAuth Google 
+google.client_id = 1001078541512-r65d0gladmatsh9bhnevhe24f6hv1ikt.apps.googleusercontent.com
+google.client_secret = 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
+
+#
+# Comptes autorises a se connecter (liste blanche verifiee en base) et comptes
+# qui portent le groupe 'group:administrators' (droit 'manage').
+auth.admin_emails = phuoc@caotek.fr
+
+# Le cookie de session porte l'identite : en local on passe en http.
+session.cookie_secure = false
+
 # By default, the toolbar only appears for clients from IP addresses
 # '127.0.0.1' and '::1'.
 # debugtoolbar.hosts = 127.0.0.1 ::1

production.ini

@@ -16,6 +16,15 @@ sqlalchemy.url = sqlite:///%(here)s/ctp_blogr.sqlite
 
 retry.attempts = 3
 
+# Les identifiants OAuth Google 
+google.client_id = 1001078541512-r65d0gladmatsh9bhnevhe24f6hv1ikt.apps.googleusercontent.com
+google.client_secret = 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
+
+auth.admin_emails = phuoc@caotek.fr
+
+# Le cookie de session porte l'identite : url_scheme est https, donc Secure.
+session.cookie_secure = true
+
 [pshell]
 setup = ctp_blogr.pshell.setup
 

setup.py

@@ -21,8 +21,8 @@ requires = [
     'zope.sqlalchemy',
     'wtforms',  # form library 2.2.1
     'webhelpers2',  # various web building related helpers 2.0
-    'passlib',
     'markdown',
+    'authlib',  # client OAuth 2.0 / OIDC pour l'authentification Google
 ]
 
 tests_require = [