SSF: handle read-only user stores when toggling ssf.notify attributes

Auto-notify-on-login (and the subject-management endpoints) wrote the
ssf.notify.<clientId> attribute unconditionally, which threw a
ReadOnlyException for users backed by a read-only LDAP federation with
import disabled — surfacing as a per-login ERROR and failing to subscribe
the user.

- Guard the ssf.notify / tombstone writes (user + org) so they only run
  when the stored value would actually change; redundant calls are now
  no-ops instead of failing on read-only stores.
- autoNotifyOnLogin catches ReadOnlyException (WARN + skip) so a read-only
  user no longer disrupts login; non-ReadOnlyException still propagates.
- Subject-management API returns SUBJECT_READ_ONLY (409) instead of a 500
  when the subject is backed by a read-only store.
- Add unit tests for the write guards and the listener's read-only handling.

Fixes #49250

Signed-off-by: Thomas Darimont <thomas.darimont@googlemail.com>

ssf/transmitter/src/main/java/org/keycloak/ssf/transmitter/event/SsfTransmitterEventListener.java

@@ -25,6 +25,7 @@ import org.keycloak.ssf.transmitter.stream.StreamService;
 import org.keycloak.ssf.transmitter.stream.storage.client.ClientStreamStore;
 import org.keycloak.ssf.transmitter.subject.SsfNotifyAttributes;
 import org.keycloak.ssf.transmitter.support.SsfUtil;
+import org.keycloak.storage.ReadOnlyException;
 
 import org.jboss.logging.Logger;
 
@@ -231,9 +232,23 @@ public class SsfTransmitterEventListener implements EventListenerProvider {
         // org doesn't need a redundant per-user attribute.
         if (!isUserNotified(transmitter, user, client)
                 && !isAnyOrganizationNotified(transmitter, user, client)) {
-            markAsNotified(user, client);
-            log.debugf("SSF auto-notify on login: tagged user %s for receiver client %s",
-                    user.getId(), client.getClientId());
+            try {
+                markAsNotified(user, client);
+                log.debugf("SSF auto-notify on login: tagged user %s for receiver client %s",
+                        user.getId(), client.getClientId());
+            } catch (ReadOnlyException e) {
+                // Read-only user store (e.g. LDAP edit mode READ_ONLY, or
+                // import disabled): the ssf.notify.<clientId> attribute
+                // can't be persisted. Degrade quietly so login isn't
+                // disrupted and we don't surface a per-login ERROR via the
+                // event framework. Auto-notify-on-login requires writable
+                // user storage (e.g. LDAP edit mode UNSYNCED with import
+                // enabled). Alterantively users can use an LDAP attribute
+                // mapper that returns a proper ssf.notify.<client_id>=true attribute
+                log.warnf("SSF auto-notify on login: cannot tag read-only user %s for receiver client %s; "
+                        + "auto-notify-on-login requires writable user storage. Skipping.",
+                        user.getId(), client.getClientId());
+            }
         }
     }
 

ssf/transmitter/src/main/java/org/keycloak/ssf/transmitter/resources/SsfSubjectManagementResource.java

@@ -109,6 +109,10 @@ public class SsfSubjectManagementResource {
             case FORMAT_UNSUPPORTED -> Response.status(Response.Status.BAD_REQUEST)
                     .entity(new SsfErrorRepresentation("invalid_request", "unsupported subject format"))
                     .build();
+            case SUBJECT_READ_ONLY -> Response.status(Response.Status.CONFLICT)
+                    .entity(new SsfErrorRepresentation("subject_read_only",
+                            "subject is backed by a read-only user store and cannot be persisted"))
+                    .build();
             default -> okEmptyJson();
         };
     }
@@ -176,6 +180,10 @@ public class SsfSubjectManagementResource {
             case FORMAT_UNSUPPORTED -> Response.status(Response.Status.BAD_REQUEST)
                     .entity(new SsfErrorRepresentation("invalid_request", "unsupported subject format"))
                     .build();
+            case SUBJECT_READ_ONLY -> Response.status(Response.Status.CONFLICT)
+                    .entity(new SsfErrorRepresentation("subject_read_only",
+                            "subject is backed by a read-only user store and cannot be persisted"))
+                    .build();
             default -> Response.noContent().build();
         };
     }

ssf/transmitter/src/main/java/org/keycloak/ssf/transmitter/subject/SsfNotifyAttributes.java

@@ -58,15 +58,29 @@ public final class SsfNotifyAttributes {
     // -- user --
 
     public static void setForUser(UserModel user, String clientId) {
-        user.setSingleAttribute(attributeKey(clientId), ATTRIBUTE_VALUE_TRUE);
+        String attributeKey = attributeKey(clientId);
+        if (!Boolean.parseBoolean(user.getFirstAttribute(attributeKey))) {
+            // only write the attribute if it's not already set or is set to false
+            user.setSingleAttribute(attributeKey, ATTRIBUTE_VALUE_TRUE);
+        }
     }
 
     public static void excludeForUser(UserModel user, String clientId) {
-        user.setSingleAttribute(attributeKey(clientId), ATTRIBUTE_VALUE_FALSE);
+        String attributeKey = attributeKey(clientId);
+        if (!ATTRIBUTE_VALUE_FALSE.equals(user.getFirstAttribute(attributeKey))) {
+            // write the exclude marker unless it is already false — an unset
+            // subject must still get an explicit "false" so the exclusion
+            // overrides a default_subjects=ALL stream or an org notify=true
+            user.setSingleAttribute(attributeKey, ATTRIBUTE_VALUE_FALSE);
+        }
     }
 
     public static void clearForUser(UserModel user, String clientId) {
-        user.removeAttribute(attributeKey(clientId));
+        String attributeKey = attributeKey(clientId);
+        if (user.getFirstAttribute(attributeKey) != null) {
+            // only clear the attribute if it's set
+            user.removeAttribute(attributeKey);
+        }
     }
 
     public static boolean isUserNotified(UserModel user, String clientId) {
@@ -104,7 +118,11 @@ public final class SsfNotifyAttributes {
     }
 
     public static void clearRemovedAtForUser(UserModel user, String clientId) {
-        user.removeAttribute(removedAtKey(clientId));
+        String removedAtKey = removedAtKey(clientId);
+        if (user.getFirstAttribute(removedAtKey) != null) {
+            // only clear the tombstone if it's set
+            user.removeAttribute(removedAtKey);
+        }
     }
 
     /**
@@ -134,20 +152,33 @@ public final class SsfNotifyAttributes {
     // -- organization --
 
     public static void setForOrganization(OrganizationModel org, String clientId) {
+        if (isOrganizationNotified(org, clientId)) {
+            // already notified — skip the redundant attribute write
+            return;
+        }
         Map<String, List<String>> attrs = new HashMap<>(org.getAttributes());
         attrs.put(attributeKey(clientId), List.of(ATTRIBUTE_VALUE_TRUE));
         org.setAttributes(attrs);
     }
 
     public static void excludeForOrganization(OrganizationModel org, String clientId) {
+        if (isOrganizationExcluded(org, clientId)) {
+            // already excluded — skip the redundant attribute write
+            return;
+        }
         Map<String, List<String>> attrs = new HashMap<>(org.getAttributes());
         attrs.put(attributeKey(clientId), List.of(ATTRIBUTE_VALUE_FALSE));
         org.setAttributes(attrs);
     }
 
     public static void clearForOrganization(OrganizationModel org, String clientId) {
+        String attributeKey = attributeKey(clientId);
+        if (!org.getAttributes().containsKey(attributeKey)) {
+            // nothing to clear
+            return;
+        }
         Map<String, List<String>> attrs = new HashMap<>(org.getAttributes());
-        attrs.remove(attributeKey(clientId));
+        attrs.remove(attributeKey);
         org.setAttributes(attrs);
     }
 
@@ -181,8 +212,13 @@ public final class SsfNotifyAttributes {
     }
 
     public static void clearRemovedAtForOrganization(OrganizationModel org, String clientId) {
+        String removedAtKey = removedAtKey(clientId);
+        if (!org.getAttributes().containsKey(removedAtKey)) {
+            // nothing to clear
+            return;
+        }
         Map<String, List<String>> attrs = new HashMap<>(org.getAttributes());
-        attrs.remove(removedAtKey(clientId));
+        attrs.remove(removedAtKey);
         org.setAttributes(attrs);
     }
 

ssf/transmitter/src/main/java/org/keycloak/ssf/transmitter/subject/SubjectManagementResult.java

@@ -4,5 +4,17 @@ public enum SubjectManagementResult {
     OK,
     STREAM_NOT_FOUND,
     FORMAT_UNSUPPORTED,
-    SUBJECT_NOT_FOUND
+    SUBJECT_NOT_FOUND,
+    /**
+     * The resolved subject is backed by a read-only user store
+     * (e.g. an LDAP federation with edit mode {@code READ_ONLY}, or
+     * with import disabled) so the {@code ssf.notify.<clientId>}
+     * subscription state cannot be persisted. Surfaces as a clean
+     * transmitter limitation rather than an unhandled 500. Persisting
+     * per-user subscription state requires writable user storage
+     * (e.g. LDAP edit mode {@code UNSYNCED} with import enabled,
+     * or a LDAP attribute mapper that returns a proper
+     * {@code ssf.notify.<client_id>=true} attribute).
+     */
+    SUBJECT_READ_ONLY
 }

ssf/transmitter/src/main/java/org/keycloak/ssf/transmitter/subject/SubjectManagementService.java

@@ -19,6 +19,7 @@ import org.keycloak.ssf.transmitter.resources.AddSubjectRequest;
 import org.keycloak.ssf.transmitter.resources.RemoveSubjectRequest;
 import org.keycloak.ssf.transmitter.stream.StreamConfig;
 import org.keycloak.ssf.transmitter.stream.storage.client.ClientStreamStore;
+import org.keycloak.storage.ReadOnlyException;
 
 import org.jboss.logging.Logger;
 
@@ -56,8 +57,12 @@ public class SubjectManagementService {
             // clear the SSF §9.3 tombstone so the dispatcher uses the
             // fresh include marker instead of falling through to a
             // stale grace-window check.
-            SsfNotifyAttributes.clearRemovedAtForUser(u.user(), callerClientId);
-            SsfNotifyAttributes.setForUser(u.user(), callerClientId);
+            try {
+                SsfNotifyAttributes.clearRemovedAtForUser(u.user(), callerClientId);
+                SsfNotifyAttributes.setForUser(u.user(), callerClientId);
+            } catch (ReadOnlyException e) {
+                return readOnlySubject("add", callerClientId, u.user());
+            }
             log.debugf("SSF subject added. clientId=%s userId=%s", callerClientId, u.user().getId());
             return SubjectManagementResult.OK;
         }
@@ -94,8 +99,12 @@ public class SubjectManagementService {
             // Explicit exclusion is an admin-trusted action — no grace
             // window. Clear any prior receiver-driven tombstone so the
             // exclude marker takes effect immediately.
-            SsfNotifyAttributes.clearRemovedAtForUser(u.user(), callerClientId);
-            SsfNotifyAttributes.excludeForUser(u.user(), callerClientId);
+            try {
+                SsfNotifyAttributes.clearRemovedAtForUser(u.user(), callerClientId);
+                SsfNotifyAttributes.excludeForUser(u.user(), callerClientId);
+            } catch (ReadOnlyException e) {
+                return readOnlySubject("exclude", callerClientId, u.user());
+            }
             log.debugf("SSF subject excluded. clientId=%s userId=%s", callerClientId, u.user().getId());
             return SubjectManagementResult.OK;
         }
@@ -123,10 +132,14 @@ public class SubjectManagementService {
      */
     protected SubjectManagementResult unregisterSubjectForNotification(String callerClientId, SubjectResolution resolution, boolean applyTombstone) {
         if (resolution instanceof SubjectResolution.User u) {
-            if (applyTombstone) {
-                SsfNotifyAttributes.stampRemovedAtForUser(u.user(), callerClientId);
+            try {
+                if (applyTombstone) {
+                    SsfNotifyAttributes.stampRemovedAtForUser(u.user(), callerClientId);
+                }
+                SsfNotifyAttributes.clearForUser(u.user(), callerClientId);
+            } catch (ReadOnlyException e) {
+                return readOnlySubject("remove", callerClientId, u.user());
             }
-            SsfNotifyAttributes.clearForUser(u.user(), callerClientId);
             log.debugf("SSF subject removed. clientId=%s userId=%s tombstone=%s",
                     callerClientId, u.user().getId(), applyTombstone);
             return SubjectManagementResult.OK;
@@ -146,6 +159,23 @@ public class SubjectManagementService {
         return SubjectManagementResult.FORMAT_UNSUPPORTED;
     }
 
+    /**
+     * Logs a read-only user store and maps it to
+     * {@link SubjectManagementResult#SUBJECT_READ_ONLY}. The
+     * {@code ssf.notify.<clientId>} subscription state can't be persisted
+     * for subjects backed by a read-only provider (e.g. LDAP edit mode
+     * {@code READ_ONLY}, or import disabled), so the caller surfaces this
+     * as a clean limitation rather than letting the {@link ReadOnlyException}
+     * escape as a 500. Organizations are always Keycloak-managed, so only
+     * the user branches need this guard.
+     */
+    protected SubjectManagementResult readOnlySubject(String operation, String callerClientId, UserModel user) {
+        log.debugf("SSF subject %s: user %s is read-only; ssf.notify state cannot be persisted. "
+                + "This requires writable user storage. clientId=%s",
+                operation, user.getId(), callerClientId);
+        return SubjectManagementResult.SUBJECT_READ_ONLY;
+    }
+
     /**
      * Resolves a {@link SubjectId} to a Keycloak entity. Protected so
      * subclasses can plug in custom resolution logic — e.g. additional

ssf/transmitter/src/test/java/org/keycloak/ssf/transmitter/event/SsfTransmitterEventListenerTest.java

@@ -3,16 +3,31 @@ package org.keycloak.ssf.transmitter.event;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.keycloak.events.Event;
+import org.keycloak.models.ClientModel;
+import org.keycloak.models.KeycloakContext;
+import org.keycloak.models.KeycloakSession;
+import org.keycloak.models.RealmModel;
+import org.keycloak.models.UserModel;
+import org.keycloak.models.UserProvider;
 import org.keycloak.ssf.event.caep.CaepCredentialChange;
 import org.keycloak.ssf.event.caep.CaepSessionRevoked;
 import org.keycloak.ssf.event.token.SsfSecurityEventToken;
+import org.keycloak.ssf.transmitter.SsfTransmitterProvider;
 import org.keycloak.ssf.transmitter.stream.StreamConfig;
+import org.keycloak.ssf.transmitter.stream.storage.client.ClientStreamStore;
+import org.keycloak.storage.ReadOnlyException;
 
 import org.junit.jupiter.api.Test;
 
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 /**
  * Unit tests for the {@code ssf.emitOnlyEvents} per-receiver gate
@@ -125,4 +140,99 @@ class SsfTransmitterEventListenerTest {
         }
         throw new IllegalArgumentException("Test fixture only covers CAEP session-revoked / credential-change. Add a branch when extending.");
     }
+
+    // ----- auto-notify-on-login read-only handling -----
+
+    private static final String RECEIVER_CLIENT_ID = "receiver";
+    private static final String LOGIN_USER_ID = "user-1";
+
+    @Test
+    void autoNotifyOnLogin_swallowsReadOnlyException() {
+        KeycloakSession session = mockLoginSession();
+        SsfTransmitterProvider transmitter = mock(SsfTransmitterProvider.class);
+        AtomicBoolean attempted = new AtomicBoolean(false);
+
+        // A user backed by a read-only store (e.g. LDAP edit mode READ_ONLY,
+        // or import disabled): markAsNotified throws ReadOnlyException. The
+        // listener must swallow it so login isn't disrupted.
+        SsfTransmitterEventListener readOnlyListener = new SsfTransmitterEventListener(session) {
+            @Override
+            protected boolean isUserNotified(SsfTransmitterProvider t, UserModel u, ClientModel c) {
+                return false;
+            }
+            @Override
+            protected boolean isAnyOrganizationNotified(SsfTransmitterProvider t, UserModel u, ClientModel c) {
+                return false;
+            }
+            @Override
+            protected void markAsNotified(UserModel u, ClientModel c) {
+                attempted.set(true);
+                throw new ReadOnlyException("user is read-only");
+            }
+        };
+
+        assertDoesNotThrow(() -> readOnlyListener.autoNotifyOnLogin(loginEvent(), transmitter),
+                "a read-only user store must not break login — the write is skipped, not propagated");
+        assertTrue(attempted.get(),
+                "the listener should have attempted the write before swallowing the exception");
+    }
+
+    @Test
+    void autoNotifyOnLogin_propagatesNonReadOnlyExceptions() {
+        KeycloakSession session = mockLoginSession();
+        SsfTransmitterProvider transmitter = mock(SsfTransmitterProvider.class);
+
+        // The catch is narrow: only ReadOnlyException is swallowed. Any
+        // other failure must still surface so genuine bugs aren't hidden.
+        SsfTransmitterEventListener faultyListener = new SsfTransmitterEventListener(session) {
+            @Override
+            protected boolean isUserNotified(SsfTransmitterProvider t, UserModel u, ClientModel c) {
+                return false;
+            }
+            @Override
+            protected boolean isAnyOrganizationNotified(SsfTransmitterProvider t, UserModel u, ClientModel c) {
+                return false;
+            }
+            @Override
+            protected void markAsNotified(UserModel u, ClientModel c) {
+                throw new IllegalStateException("boom");
+            }
+        };
+
+        assertThrows(IllegalStateException.class,
+                () -> faultyListener.autoNotifyOnLogin(loginEvent(), transmitter),
+                "only ReadOnlyException is swallowed; other failures must surface");
+    }
+
+    /**
+     * Wires a session whose realm hosts an SSF receiver client with
+     * {@code ssf.enabled=true} and {@code ssf.autoNotifyOnLogin=true},
+     * resolving {@link #LOGIN_USER_ID} to a (mock) user — enough for
+     * {@code autoNotifyOnLogin} to reach the {@code markAsNotified} write.
+     */
+    private KeycloakSession mockLoginSession() {
+        KeycloakSession session = mock(KeycloakSession.class);
+        KeycloakContext context = mock(KeycloakContext.class);
+        RealmModel realm = mock(RealmModel.class);
+        ClientModel client = mock(ClientModel.class);
+        UserModel user = mock(UserModel.class);
+        UserProvider users = mock(UserProvider.class);
+
+        when(session.getContext()).thenReturn(context);
+        when(context.getRealm()).thenReturn(realm);
+        when(realm.getClientByClientId(RECEIVER_CLIENT_ID)).thenReturn(client);
+        when(client.getAttribute(ClientStreamStore.SSF_ENABLED_KEY)).thenReturn("true");
+        when(client.getAttribute(ClientStreamStore.SSF_AUTO_NOTIFY_ON_LOGIN_KEY)).thenReturn("true");
+        // SSF_DEFAULT_SUBJECTS_KEY left unstubbed (null) → not broadcast (ALL).
+        when(session.users()).thenReturn(users);
+        when(users.getUserById(realm, LOGIN_USER_ID)).thenReturn(user);
+        return session;
+    }
+
+    private Event loginEvent() {
+        Event event = new Event();
+        event.setClientId(RECEIVER_CLIENT_ID);
+        event.setUserId(LOGIN_USER_ID);
+        return event;
+    }
 }

ssf/transmitter/src/test/java/org/keycloak/ssf/transmitter/subject/SsfNotifyAttributesTest.java

@@ -1,15 +1,37 @@
 package org.keycloak.ssf.transmitter.subject;
 
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.keycloak.models.OrganizationModel;
+import org.keycloak.models.UserModel;
+
 import org.junit.jupiter.api.Test;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 /**
- * Unit tests for {@link SsfNotifyAttributes} attribute key derivation
- * and value semantics.
+ * Unit tests for {@link SsfNotifyAttributes} attribute key derivation,
+ * value semantics, and the write guards that keep the
+ * {@code ssf.notify.<clientId>} / tombstone attributes from being
+ * rewritten when their value would not change. The guards matter for
+ * read-only user stores (e.g. LDAP edit mode {@code READ_ONLY}, or with
+ * import disabled), where a redundant write would throw a
+ * {@code ReadOnlyException} instead of being a harmless no-op.
  */
 class SsfNotifyAttributesTest {
 
+    private static final String CLIENT_ID = "rcv";
+    private static final String NOTIFY_KEY = "ssf.notify.rcv";
+    private static final String REMOVED_AT_KEY = "ssf.notifyRemovedAt.rcv";
+
     @Test
     void attributeKey_prefixesClientId() {
         assertEquals("ssf.notify.abc-123", SsfNotifyAttributes.attributeKey("abc-123"));
@@ -25,4 +47,205 @@ class SsfNotifyAttributesTest {
     void attributePrefix_isConsistent() {
         assertEquals("ssf.notify.", SsfNotifyAttributes.ATTRIBUTE_PREFIX);
     }
+
+    @Test
+    void attributeKey_handlesUrlClientId() {
+        // Receiver OAuth client_ids may be URLs.
+        String clientId = "https://rp.example.com";
+        assertEquals("ssf.notify.https://rp.example.com",
+                SsfNotifyAttributes.attributeKey(clientId));
+        assertEquals("ssf.notifyRemovedAt.https://rp.example.com",
+                SsfNotifyAttributes.removedAtKey(clientId));
+    }
+
+    // ----- user: setForUser -----
+
+    @Test
+    void setForUser_writesTrue_whenUnset() {
+        UserModel user = mock(UserModel.class);
+        when(user.getFirstAttribute(NOTIFY_KEY)).thenReturn(null);
+
+        SsfNotifyAttributes.setForUser(user, CLIENT_ID);
+
+        verify(user).setSingleAttribute(NOTIFY_KEY, "true");
+    }
+
+    @Test
+    void setForUser_writesTrue_whenCurrentlyFalse() {
+        UserModel user = mock(UserModel.class);
+        when(user.getFirstAttribute(NOTIFY_KEY)).thenReturn("false");
+
+        SsfNotifyAttributes.setForUser(user, CLIENT_ID);
+
+        verify(user).setSingleAttribute(NOTIFY_KEY, "true");
+    }
+
+    @Test
+    void setForUser_skips_whenAlreadyTrue() {
+        UserModel user = mock(UserModel.class);
+        when(user.getFirstAttribute(NOTIFY_KEY)).thenReturn("true");
+
+        SsfNotifyAttributes.setForUser(user, CLIENT_ID);
+
+        verify(user, never()).setSingleAttribute(anyString(), anyString());
+    }
+
+    // ----- user: excludeForUser -----
+
+    @Test
+    void excludeForUser_writesFalse_whenCurrentlyTrue() {
+        UserModel user = mock(UserModel.class);
+        when(user.getFirstAttribute(NOTIFY_KEY)).thenReturn("true");
+
+        SsfNotifyAttributes.excludeForUser(user, CLIENT_ID);
+
+        verify(user).setSingleAttribute(NOTIFY_KEY, "false");
+    }
+
+    @Test
+    void excludeForUser_writesFalse_whenUnset() {
+        // Ignoring an unset subject must write an explicit "false" so the
+        // exclusion actually takes effect — e.g. to override a
+        // default_subjects=ALL stream or an org-level notify=true.
+        UserModel user = mock(UserModel.class);
+        when(user.getFirstAttribute(NOTIFY_KEY)).thenReturn(null);
+
+        SsfNotifyAttributes.excludeForUser(user, CLIENT_ID);
+
+        verify(user).setSingleAttribute(NOTIFY_KEY, "false");
+    }
+
+    @Test
+    void excludeForUser_skips_whenAlreadyFalse() {
+        UserModel user = mock(UserModel.class);
+        when(user.getFirstAttribute(NOTIFY_KEY)).thenReturn("false");
+
+        SsfNotifyAttributes.excludeForUser(user, CLIENT_ID);
+
+        verify(user, never()).setSingleAttribute(anyString(), anyString());
+    }
+
+    // ----- user: clearForUser / clearRemovedAtForUser -----
+
+    @Test
+    void clearForUser_removes_whenSet() {
+        UserModel user = mock(UserModel.class);
+        when(user.getFirstAttribute(NOTIFY_KEY)).thenReturn("true");
+
+        SsfNotifyAttributes.clearForUser(user, CLIENT_ID);
+
+        verify(user).removeAttribute(NOTIFY_KEY);
+    }
+
+    @Test
+    void clearForUser_skips_whenAbsent() {
+        UserModel user = mock(UserModel.class);
+        when(user.getFirstAttribute(NOTIFY_KEY)).thenReturn(null);
+
+        SsfNotifyAttributes.clearForUser(user, CLIENT_ID);
+
+        verify(user, never()).removeAttribute(anyString());
+    }
+
+    @Test
+    void clearRemovedAtForUser_removes_whenSet() {
+        UserModel user = mock(UserModel.class);
+        when(user.getFirstAttribute(REMOVED_AT_KEY)).thenReturn("123");
+
+        SsfNotifyAttributes.clearRemovedAtForUser(user, CLIENT_ID);
+
+        verify(user).removeAttribute(REMOVED_AT_KEY);
+    }
+
+    @Test
+    void clearRemovedAtForUser_skips_whenAbsent() {
+        UserModel user = mock(UserModel.class);
+        when(user.getFirstAttribute(REMOVED_AT_KEY)).thenReturn(null);
+
+        SsfNotifyAttributes.clearRemovedAtForUser(user, CLIENT_ID);
+
+        verify(user, never()).removeAttribute(anyString());
+    }
+
+    // ----- organization: setForOrganization -----
+
+    @Test
+    void setForOrganization_writesTrue_whenUnset() {
+        OrganizationModel org = mock(OrganizationModel.class);
+        when(org.getAttributes()).thenReturn(new HashMap<>());
+
+        SsfNotifyAttributes.setForOrganization(org, CLIENT_ID);
+
+        verify(org).setAttributes(Map.of(NOTIFY_KEY, List.of("true")));
+    }
+
+    @Test
+    void setForOrganization_skips_whenAlreadyNotified() {
+        OrganizationModel org = mock(OrganizationModel.class);
+        when(org.getAttributes()).thenReturn(Map.of(NOTIFY_KEY, List.of("true")));
+
+        SsfNotifyAttributes.setForOrganization(org, CLIENT_ID);
+
+        verify(org, never()).setAttributes(anyMap());
+    }
+
+    // ----- organization: excludeForOrganization -----
+
+    @Test
+    void excludeForOrganization_writesFalse_whenUnset() {
+        // Org exclude differs from user exclude: it writes an explicit
+        // "false" even when unset, so a broadcast (default_subjects=ALL)
+        // org can be opted out.
+        OrganizationModel org = mock(OrganizationModel.class);
+        when(org.getAttributes()).thenReturn(new HashMap<>());
+
+        SsfNotifyAttributes.excludeForOrganization(org, CLIENT_ID);
+
+        verify(org).setAttributes(Map.of(NOTIFY_KEY, List.of("false")));
+    }
+
+    @Test
+    void excludeForOrganization_skips_whenAlreadyExcluded() {
+        OrganizationModel org = mock(OrganizationModel.class);
+        when(org.getAttributes()).thenReturn(Map.of(NOTIFY_KEY, List.of("false")));
+
+        SsfNotifyAttributes.excludeForOrganization(org, CLIENT_ID);
+
+        verify(org, never()).setAttributes(anyMap());
+    }
+
+    // ----- organization: clearForOrganization / clearRemovedAtForOrganization -----
+
+    @Test
+    void clearForOrganization_removes_whenPresent_andKeepsOtherAttributes() {
+        OrganizationModel org = mock(OrganizationModel.class);
+        Map<String, List<String>> attrs = new HashMap<>();
+        attrs.put(NOTIFY_KEY, List.of("true"));
+        attrs.put("unrelated", List.of("keep"));
+        when(org.getAttributes()).thenReturn(attrs);
+
+        SsfNotifyAttributes.clearForOrganization(org, CLIENT_ID);
+
+        verify(org).setAttributes(Map.of("unrelated", List.of("keep")));
+    }
+
+    @Test
+    void clearForOrganization_skips_whenAbsent() {
+        OrganizationModel org = mock(OrganizationModel.class);
+        when(org.getAttributes()).thenReturn(new HashMap<>());
+
+        SsfNotifyAttributes.clearForOrganization(org, CLIENT_ID);
+
+        verify(org, never()).setAttributes(anyMap());
+    }
+
+    @Test
+    void clearRemovedAtForOrganization_skips_whenAbsent() {
+        OrganizationModel org = mock(OrganizationModel.class);
+        when(org.getAttributes()).thenReturn(new HashMap<>());
+
+        SsfNotifyAttributes.clearRemovedAtForOrganization(org, CLIENT_ID);
+
+        verify(org, never()).setAttributes(anyMap());
+    }
 }