WIP: Auth in add dialog

app/schemas/io.heckel.ntfy.db.Database/7.json

@@ -2,11 +2,11 @@
   "formatVersion": 1,
   "database": {
     "version": 7,
-    "identityHash": "ecb1b85b2ae822dc62b2843620368477",
+    "identityHash": "12fd7305f39828bf44164435d48b7e56",
     "entities": [
       {
         "tableName": "Subscription",
-        "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER NOT NULL, `baseUrl` TEXT NOT NULL, `topic` TEXT NOT NULL, `instant` INTEGER NOT NULL, `mutedUntil` INTEGER NOT NULL, `upAppId` TEXT, `upConnectorToken` TEXT, PRIMARY KEY(`id`))",
+        "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER NOT NULL, `baseUrl` TEXT NOT NULL, `topic` TEXT NOT NULL, `instant` INTEGER NOT NULL, `mutedUntil` INTEGER NOT NULL, `authUserId` INTEGER, `upAppId` TEXT, `upConnectorToken` TEXT, PRIMARY KEY(`id`))",
         "fields": [
           {
             "fieldPath": "id",
@@ -38,6 +38,12 @@
             "affinity": "INTEGER",
             "notNull": true
           },
+          {
+            "fieldPath": "authUserId",
+            "columnName": "authUserId",
+            "affinity": "INTEGER",
+            "notNull": false
+          },
           {
             "fieldPath": "upAppId",
             "columnName": "upAppId",
@@ -65,6 +71,7 @@
               "baseUrl",
               "topic"
             ],
+            "orders": [],
             "createSql": "CREATE UNIQUE INDEX IF NOT EXISTS `index_Subscription_baseUrl_topic` ON `${TABLE_NAME}` (`baseUrl`, `topic`)"
           },
           {
@@ -73,6 +80,7 @@
             "columnNames": [
               "upConnectorToken"
             ],
+            "orders": [],
             "createSql": "CREATE UNIQUE INDEX IF NOT EXISTS `index_Subscription_upConnectorToken` ON `${TABLE_NAME}` (`upConnectorToken`)"
           }
         ],
@@ -196,6 +204,38 @@
         "indices": [],
         "foreignKeys": []
       },
+      {
+        "tableName": "User",
+        "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `username` TEXT NOT NULL, `password` TEXT NOT NULL)",
+        "fields": [
+          {
+            "fieldPath": "id",
+            "columnName": "id",
+            "affinity": "INTEGER",
+            "notNull": true
+          },
+          {
+            "fieldPath": "username",
+            "columnName": "username",
+            "affinity": "TEXT",
+            "notNull": true
+          },
+          {
+            "fieldPath": "password",
+            "columnName": "password",
+            "affinity": "TEXT",
+            "notNull": true
+          }
+        ],
+        "primaryKey": {
+          "columnNames": [
+            "id"
+          ],
+          "autoGenerate": true
+        },
+        "indices": [],
+        "foreignKeys": []
+      },
       {
         "tableName": "Log",
         "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `timestamp` INTEGER NOT NULL, `tag` TEXT NOT NULL, `level` INTEGER NOT NULL, `message` TEXT NOT NULL, `exception` TEXT)",
@@ -250,7 +290,7 @@
     "views": [],
     "setupQueries": [
       "CREATE TABLE IF NOT EXISTS room_master_table (id INTEGER PRIMARY KEY,identity_hash TEXT)",
-      "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, 'ecb1b85b2ae822dc62b2843620368477')"
+      "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, '12fd7305f39828bf44164435d48b7e56')"
     ]
   }
 }

app/src/main/java/io/heckel/ntfy/app/Application.kt

@@ -13,7 +13,7 @@ class Application : Application() {
     }
     val repository by lazy {
         val sharedPrefs = applicationContext.getSharedPreferences(Repository.SHARED_PREFS_ID, Context.MODE_PRIVATE)
-        val repository = Repository.getInstance(sharedPrefs, database.subscriptionDao(), database.notificationDao())
+        val repository = Repository.getInstance(sharedPrefs, database)
         if (repository.getRecordLogs()) {
             Log.setRecord(true)
         }

app/src/main/java/io/heckel/ntfy/db/Database.kt

@@ -13,6 +13,7 @@ data class Subscription(
     @ColumnInfo(name = "topic") val topic: String,
     @ColumnInfo(name = "instant") val instant: Boolean,
     @ColumnInfo(name = "mutedUntil") val mutedUntil: Long, // TODO notificationSound, notificationSchedule
+    @ColumnInfo(name = "authUserId") val authUserId: Long?,
     @ColumnInfo(name = "upAppId") val upAppId: String?, // UnifiedPush application package name
     @ColumnInfo(name = "upConnectorToken") val upConnectorToken: String?, // UnifiedPush connector token
     // TODO autoDownloadAttachments, minPriority
@@ -21,8 +22,8 @@ data class Subscription(
     @Ignore val lastActive: Long = 0, // Unix timestamp
     @Ignore val state: ConnectionState = ConnectionState.NOT_APPLICABLE
 ) {
-    constructor(id: Long, baseUrl: String, topic: String, instant: Boolean, mutedUntil: Long, upAppId: String, upConnectorToken: String) :
+    constructor(id: Long, baseUrl: String, topic: String, instant: Boolean, mutedUntil: Long, authUserId: Long?, upAppId: String, upConnectorToken: String) :
-            this(id, baseUrl, topic, instant, mutedUntil, upAppId, upConnectorToken, 0, 0, 0, ConnectionState.NOT_APPLICABLE)
+            this(id, baseUrl, topic, instant, mutedUntil, authUserId, upAppId, upConnectorToken, 0, 0, 0, ConnectionState.NOT_APPLICABLE)
 }
 
 enum class ConnectionState {
@@ -35,6 +36,7 @@ data class SubscriptionWithMetadata(
     val topic: String,
     val instant: Boolean,
     val mutedUntil: Long,
+    val authUserId: Long?,
     val upAppId: String?,
     val upConnectorToken: String?,
     val totalCount: Int,
@@ -77,6 +79,15 @@ const val PROGRESS_FAILED = -3
 const val PROGRESS_DELETED = -4
 const val PROGRESS_DONE = 100
 
+@Entity
+data class User(
+    @PrimaryKey(autoGenerate = true) val id: Long,
+    @ColumnInfo(name = "username") val username: String,
+    @ColumnInfo(name = "password") val password: String
+) {
+    override fun toString(): String = username
+}
+
 @Entity(tableName = "Log")
 data class LogEntry(
     @PrimaryKey(autoGenerate = true) val id: Long, // Internal ID, only used in Repository and activities
@@ -90,10 +101,11 @@ data class LogEntry(
             this(0, timestamp, tag, level, message, exception)
 }
 
-@androidx.room.Database(entities = [Subscription::class, Notification::class, LogEntry::class], version = 7)
+@androidx.room.Database(entities = [Subscription::class, Notification::class, User::class, LogEntry::class], version = 7)
 abstract class Database : RoomDatabase() {
     abstract fun subscriptionDao(): SubscriptionDao
     abstract fun notificationDao(): NotificationDao
+    abstract fun userDao(): UserDao
     abstract fun logDao(): LogDao
 
     companion object {
@@ -180,7 +192,7 @@ abstract class Database : RoomDatabase() {
 interface SubscriptionDao {
     @Query("""
         SELECT 
-          s.id, s.baseUrl, s.topic, s.instant, s.mutedUntil, s.upAppId, s.upConnectorToken,
+          s.id, s.baseUrl, s.topic, s.instant, s.mutedUntil, s.authUserId, s.upAppId, s.upConnectorToken,
           COUNT(n.id) totalCount, 
           COUNT(CASE n.notificationId WHEN 0 THEN NULL ELSE n.id END) newCount, 
           IFNULL(MAX(n.timestamp),0) AS lastActive
@@ -193,7 +205,7 @@ interface SubscriptionDao {
 
     @Query("""
         SELECT 
-          s.id, s.baseUrl, s.topic, s.instant, s.mutedUntil, s.upAppId, s.upConnectorToken,
+          s.id, s.baseUrl, s.topic, s.instant, s.mutedUntil, s.authUserId, s.upAppId, s.upConnectorToken,
           COUNT(n.id) totalCount, 
           COUNT(CASE n.notificationId WHEN 0 THEN NULL ELSE n.id END) newCount, 
           IFNULL(MAX(n.timestamp),0) AS lastActive
@@ -206,7 +218,7 @@ interface SubscriptionDao {
 
     @Query("""
         SELECT 
-          s.id, s.baseUrl, s.topic, s.instant, s.mutedUntil, s.upAppId, s.upConnectorToken,
+          s.id, s.baseUrl, s.topic, s.instant, s.mutedUntil, s.authUserId, s.upAppId, s.upConnectorToken,
           COUNT(n.id) totalCount, 
           COUNT(CASE n.notificationId WHEN 0 THEN NULL ELSE n.id END) newCount, 
           IFNULL(MAX(n.timestamp),0) AS lastActive
@@ -283,6 +295,24 @@ interface NotificationDao {
     fun removeAll(subscriptionId: Long)
 }
 
+@Dao
+interface UserDao {
+    @Insert
+    suspend fun insert(user: User)
+
+    @Query("SELECT * FROM user ORDER BY username")
+    suspend fun list(): List<User>
+
+    @Query("SELECT * FROM user WHERE id = :id")
+    suspend fun get(id: Long): User
+
+    @Update
+    suspend fun update(user: User)
+
+    @Delete
+    suspend fun delete(user: User)
+}
+
 @Dao
 interface LogDao {
     @Insert

app/src/main/java/io/heckel/ntfy/db/Repository.kt

@@ -8,11 +8,14 @@ import androidx.annotation.WorkerThread
 import androidx.appcompat.app.AppCompatDelegate
 import androidx.lifecycle.*
 import io.heckel.ntfy.log.Log
-import java.util.*
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.AtomicLong
 
-class Repository(private val sharedPrefs: SharedPreferences, private val subscriptionDao: SubscriptionDao, private val notificationDao: NotificationDao) {
+class Repository(private val sharedPrefs: SharedPreferences, private val database: Database) {
+    private val subscriptionDao = database.subscriptionDao()
+    private val notificationDao = database.notificationDao()
+    private val userDao = database.userDao()
+
     private val connectionStates = ConcurrentHashMap<Long, ConnectionState>()
     private val connectionStatesLiveData = MutableLiveData(connectionStates)
     val detailViewSubscriptionId = AtomicLong(0L) // Omg, what a hack ...
@@ -113,7 +116,6 @@ class Repository(private val sharedPrefs: SharedPreferences, private val subscri
         notificationDao.update(notification)
     }
 
-
     @Suppress("RedundantSuspendModifier")
     @WorkerThread
     suspend fun markAsDeleted(notificationId: String) {
@@ -130,6 +132,18 @@ class Repository(private val sharedPrefs: SharedPreferences, private val subscri
         notificationDao.removeAll(subscriptionId)
     }
 
+    suspend fun getUsers(): List<User> {
+        return userDao.list()
+    }
+
+    suspend fun addUser(user: User) {
+        return userDao.insert(user)
+    }
+
+    suspend fun getUser(userId: Long): User {
+        return userDao.get(userId)
+    }
+
     fun getPollWorkerVersion(): Int {
         return sharedPrefs.getInt(SHARED_PREFS_POLL_WORKER_VERSION, 0)
     }
@@ -316,6 +330,7 @@ class Repository(private val sharedPrefs: SharedPreferences, private val subscri
                 topic = s.topic,
                 instant = s.instant,
                 mutedUntil = s.mutedUntil,
+                authUserId = s.authUserId,
                 upAppId = s.upAppId,
                 upConnectorToken = s.upConnectorToken,
                 totalCount = s.totalCount,
@@ -336,6 +351,7 @@ class Repository(private val sharedPrefs: SharedPreferences, private val subscri
             topic = s.topic,
             instant = s.instant,
             mutedUntil = s.mutedUntil,
+            authUserId = s.authUserId,
             upAppId = s.upAppId,
             upConnectorToken = s.upConnectorToken,
             totalCount = s.totalCount,
@@ -403,12 +419,12 @@ class Repository(private val sharedPrefs: SharedPreferences, private val subscri
         fun getInstance(activity: Activity): Repository {
             val database = Database.getInstance(activity.applicationContext)
             val sharedPrefs = activity.getSharedPreferences(SHARED_PREFS_ID, Context.MODE_PRIVATE)
-            return getInstance(sharedPrefs, database.subscriptionDao(), database.notificationDao())
+            return getInstance(sharedPrefs, database)
         }
 
-        fun getInstance(sharedPrefs: SharedPreferences, subscriptionDao: SubscriptionDao, notificationDao: NotificationDao): Repository {
+        fun getInstance(sharedPrefs: SharedPreferences, database: Database): Repository {
             return synchronized(Repository::class) {
-                val newInstance = instance ?: Repository(sharedPrefs, subscriptionDao, notificationDao)
+                val newInstance = instance ?: Repository(sharedPrefs, database)
                 instance = newInstance
                 newInstance
             }

app/src/main/java/io/heckel/ntfy/msg/ApiService.kt

@@ -3,14 +3,21 @@ package io.heckel.ntfy.msg
 import android.os.Build
 import io.heckel.ntfy.BuildConfig
 import io.heckel.ntfy.db.Notification
+import io.heckel.ntfy.db.User
 import io.heckel.ntfy.log.Log
-import io.heckel.ntfy.util.*
+import io.heckel.ntfy.util.topicUrl
+import io.heckel.ntfy.util.topicUrlAuth
+import io.heckel.ntfy.util.topicUrlJson
+import io.heckel.ntfy.util.topicUrlJsonPoll
 import okhttp3.*
 import okhttp3.RequestBody.Companion.toRequestBody
 import java.io.IOException
+import java.nio.charset.StandardCharsets
+import java.nio.charset.StandardCharsets.UTF_8
 import java.util.concurrent.TimeUnit
 import kotlin.random.Random
 
+
 class ApiService {
     private val client = OkHttpClient.Builder()
         .callTimeout(15, TimeUnit.SECONDS) // Total timeout for entire request
@@ -79,17 +86,21 @@ class ApiService {
         baseUrl: String,
         topics: String,
         since: Long,
+        user: User?,
         notify: (topic: String, Notification) -> Unit,
         fail: (Exception) -> Unit
     ): Call {
         val sinceVal = if (since == 0L) "all" else since.toString()
         val url = topicUrlJson(baseUrl, topics, sinceVal)
         Log.d(TAG, "Opening subscription connection to $url")
-
+        val builder = Request.Builder()
-        val request = Request.Builder()
+            .get()
             .url(url)
             .addHeader("User-Agent", USER_AGENT)
-            .build()
+        if (user != null) {
+            builder.addHeader("Authorization", Credentials.basic(user.username, user.password, UTF_8))
+        }
+        val request = builder.build()
         val call = subscriberClient.newCall(request)
         call.enqueue(object : Callback {
             override fun onResponse(call: Call, response: Response) {
@@ -118,6 +129,34 @@ class ApiService {
         return call
     }
 
+    fun checkAnonTopicRead(baseUrl: String, topic: String): Boolean {
+        return checkTopicRead(baseUrl, topic, creds = null)
+    }
+
+    fun checkUserTopicRead(baseUrl: String, topic: String, username: String, password: String): Boolean {
+        Log.d(TAG, "Authorizing user $username against ${topicUrl(baseUrl, topic)}")
+        return checkTopicRead(baseUrl, topic, creds = Credentials.basic(username, password, UTF_8))
+    }
+
+    private fun checkTopicRead(baseUrl: String, topic: String, creds: String?): Boolean {
+        val url = topicUrlAuth(baseUrl, topic)
+        val builder = Request.Builder()
+            .get()
+            .url(url)
+            .addHeader("User-Agent", USER_AGENT)
+        if (creds != null) {
+            builder.addHeader("Authorization", creds)
+        }
+        val request = builder.build()
+        client.newCall(request).execute().use { response ->
+            if (creds == null) {
+                return response.isSuccessful || response.code == 404 // Treat 404 as success (old server; to be removed in future versions)
+            } else {
+                return response.isSuccessful
+            }
+        }
+    }
+
     companion object {
         val USER_AGENT = "ntfy/${BuildConfig.VERSION_NAME} (${BuildConfig.FLAVOR}; Android ${Build.VERSION.RELEASE}; SDK ${Build.VERSION.SDK_INT})"
         private const val TAG = "NtfyApiService"

app/src/main/java/io/heckel/ntfy/service/Connection.kt

@@ -4,5 +4,10 @@ interface Connection {
     fun start()
     fun close()
     fun since(): Long
-    fun matches(otherSubscriptionIds: Collection<Long>): Boolean
 }
+
+data class ConnectionId(
+    val baseUrl: String,
+    val authUserId: Long?,
+    val topicsToSubscriptionIds: Map<String, Long>
+)

app/src/main/java/io/heckel/ntfy/service/JsonConnection.kt

@@ -1,9 +1,6 @@
 package io.heckel.ntfy.service
 
-import io.heckel.ntfy.db.ConnectionState
+import io.heckel.ntfy.db.*
-import io.heckel.ntfy.db.Notification
-import io.heckel.ntfy.db.Repository
-import io.heckel.ntfy.db.Subscription
 import io.heckel.ntfy.log.Log
 import io.heckel.ntfy.msg.ApiService
 import io.heckel.ntfy.util.topicUrl
@@ -12,16 +9,18 @@ import okhttp3.Call
 import java.util.concurrent.atomic.AtomicBoolean
 
 class JsonConnection(
+    private val connectionId: ConnectionId,
     private val scope: CoroutineScope,
     private val repository: Repository,
     private val api: ApiService,
-    private val baseUrl: String,
+    private val user: User?,
     private val sinceTime: Long,
-    private val topicsToSubscriptionIds: Map<String, Long>, // Topic -> Subscription ID
     private val stateChangeListener: (Collection<Long>, ConnectionState) -> Unit,
     private val notificationListener: (Subscription, Notification) -> Unit,
     private val serviceActive: () -> Boolean
 ) : Connection {
+    private val baseUrl = connectionId.baseUrl
+    private val topicsToSubscriptionIds = connectionId.topicsToSubscriptionIds
     private val subscriptionIds = topicsToSubscriptionIds.values
     private val topicsStr = topicsToSubscriptionIds.keys.joinToString(separator = ",")
     private val url = topicUrl(baseUrl, topicsStr)
@@ -57,7 +56,7 @@ class JsonConnection(
                 // Call /json subscribe endpoint and loop until the call fails, is canceled,
                 // or the job or service are cancelled/stopped
                 try {
-                    call = api.subscribe(baseUrl, topicsStr, since, notify, fail)
+                    call = api.subscribe(baseUrl, topicsStr, since, user, notify, fail)
                     while (!failed.get() && !call.isCanceled() && isActive && serviceActive()) {
                         stateChangeListener(subscriptionIds, ConnectionState.CONNECTED)
                         Log.d(TAG,"[$url] Connection is active (failed=$failed, callCanceled=${call.isCanceled()}, jobActive=$isActive, serviceStarted=${serviceActive()}")
@@ -92,10 +91,6 @@ class JsonConnection(
         if (this::call.isInitialized) call?.cancel()
     }
 
-    override fun matches(otherSubscriptionIds: Collection<Long>): Boolean {
-        return subscriptionIds.toSet() == otherSubscriptionIds.toSet()
-    }
-
     private fun nextRetryMillis(retryMillis: Long, startTime: Long): Long {
         val connectionDurationMillis = System.currentTimeMillis() - startTime
         if (connectionDurationMillis > RETRY_RESET_AFTER_MILLIS) {

app/src/main/java/io/heckel/ntfy/service/SubscriberService.kt

@@ -59,7 +59,7 @@ class SubscriberService : Service() {
     private var isServiceStarted = false
     private val repository by lazy { (application as Application).repository }
     private val dispatcher by lazy { NotificationDispatcher(this, repository) }
-    private val connections = ConcurrentHashMap<String, Connection>() // Base URL -> Connection
+    private val connections = ConcurrentHashMap<ConnectionId, Connection>()
     private val api = ApiService()
     private var notificationManager: NotificationManager? = null
     private var serviceNotification: Notification? = null
@@ -153,48 +153,56 @@ class SubscriberService : Service() {
             // Group INSTANT subscriptions by base URL, there is only one connection per base URL
             val instantSubscriptions = repository.getSubscriptions()
                 .filter { s -> s.instant }
-            val instantSubscriptionsByBaseUrl = instantSubscriptions // BaseUrl->Map[Topic->SubscriptionId]
+            val activeConnectionIds = connections.keys().toList().toSet()
-                .groupBy { s -> s.baseUrl }
+            val desiredConnectionIds = instantSubscriptions // Set<ConnectionId>
-                .mapValues { entry ->
+                .groupBy { s -> ConnectionId(s.baseUrl, s.authUserId, emptyMap()) }
-                    entry.value.associate { subscription -> subscription.topic to subscription.id }
+                .map { entry -> entry.key.copy(topicsToSubscriptionIds = entry.value.associate { s -> s.topic to s.id }) }
-                }
+                .toSet()
+            val newConnectionIds = desiredConnectionIds subtract activeConnectionIds
+            val obsoleteConnectionIds = activeConnectionIds subtract desiredConnectionIds
+            val match = activeConnectionIds == desiredConnectionIds
 
             Log.d(TAG, "Refreshing subscriptions")
-            Log.d(TAG, "- Subscriptions: $instantSubscriptionsByBaseUrl")
+            Log.d(TAG, "- Desired connections: $desiredConnectionIds")
-            Log.d(TAG, "- Active connections: $connections")
+            Log.d(TAG, "- Active connections: $activeConnectionIds")
+            Log.d(TAG, "- New connections: $newConnectionIds")
+            Log.d(TAG, "- Obsolete connections: $obsoleteConnectionIds")
+            Log.d(TAG, "- Match? --> $match")
+
+            if (match) {
+                Log.d(TAG, "- No action required.")
+                return@launch
+            }
+
+            // Open new connections
+            newConnectionIds.forEach { connectionId ->
+                // FIXME since !!!
 
-            // Start new connections and restart connections (if subscriptions have changed)
-            instantSubscriptionsByBaseUrl.forEach { (baseUrl, subscriptions) ->
                 // Do NOT request old messages for new connections; we'll call poll() in MainActivity.
                 // This is important, so we don't download attachments from old messages, which is not desired.
-                var since = System.currentTimeMillis()/1000
+
-                val connection = connections[baseUrl]
+                val since = System.currentTimeMillis()/1000
-                if (connection != null && !connection.matches(subscriptions.values)) {
-                    since = connection.since()
-                    connections.remove(baseUrl)
-                    connection.close()
-                }
-                if (!connections.containsKey(baseUrl)) {
                 val serviceActive = { -> isServiceStarted }
+                val user = if (connectionId.authUserId != null) {
+                    repository.getUser(connectionId.authUserId)
+                } else {
+                    null
+                }
                 val connection = if (repository.getConnectionProtocol() == Repository.CONNECTION_PROTOCOL_WS) {
                     val alarmManager = getSystemService(ALARM_SERVICE) as AlarmManager
-                        WsConnection(repository, baseUrl, since, subscriptions, ::onStateChanged, ::onNotificationReceived, alarmManager)
+                    WsConnection(connectionId, repository, user, since, ::onStateChanged, ::onNotificationReceived, alarmManager)
                 } else {
-                        JsonConnection(this, repository, api, baseUrl, since, subscriptions, ::onStateChanged, ::onNotificationReceived, serviceActive)
+                    JsonConnection(connectionId, this, repository, api, user, since, ::onStateChanged, ::onNotificationReceived, serviceActive)
                 }
-                    connections[baseUrl] = connection
+                connections[connectionId] = connection
                 connection.start()
             }
-            }
 
             // Close connections without subscriptions
-            val baseUrls = instantSubscriptionsByBaseUrl.keys
+            obsoleteConnectionIds.forEach { connectionId ->
-            connections.keys().toList().forEach { baseUrl ->
+                val connection = connections.remove(connectionId)
-                if (!baseUrls.contains(baseUrl)) {
-                    val connection = connections.remove(baseUrl)
                 connection?.close()
             }
-            }
 
             // Update foreground service notification popup
             if (connections.size > 0) {

app/src/main/java/io/heckel/ntfy/service/WsConnection.kt

@@ -4,15 +4,14 @@ import android.app.AlarmManager
 import android.os.Build
 import android.os.Handler
 import android.os.Looper
-import io.heckel.ntfy.db.ConnectionState
+import io.heckel.ntfy.db.*
-import io.heckel.ntfy.db.Notification
-import io.heckel.ntfy.db.Repository
-import io.heckel.ntfy.db.Subscription
 import io.heckel.ntfy.log.Log
+import io.heckel.ntfy.msg.ApiService
 import io.heckel.ntfy.msg.NotificationParser
 import io.heckel.ntfy.util.topicUrl
 import io.heckel.ntfy.util.topicUrlWs
 import okhttp3.*
+import java.nio.charset.StandardCharsets
 import java.util.*
 import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicLong
@@ -29,10 +28,10 @@ import kotlin.random.Random
  * https://github.com/gotify/android/blob/master/app/src/main/java/com/github/gotify/service/WebSocketConnection.java
  */
 class WsConnection(
+    private val connectionId: ConnectionId,
     private val repository: Repository,
-    private val baseUrl: String,
+    private val user: User?,
     private val sinceTime: Long,
-    private val topicsToSubscriptionIds: Map<String, Long>, // Topic -> Subscription ID
     private val stateChangeListener: (Collection<Long>, ConnectionState) -> Unit,
     private val notificationListener: (Subscription, Notification) -> Unit,
     private val alarmManager: AlarmManager
@@ -49,6 +48,8 @@ class WsConnection(
     private var closed = false
 
     private var since: Long = sinceTime
+    private val baseUrl = connectionId.baseUrl
+    private val topicsToSubscriptionIds = connectionId.topicsToSubscriptionIds
     private val subscriptionIds = topicsToSubscriptionIds.values
     private val topicsStr = topicsToSubscriptionIds.keys.joinToString(separator = ",")
     private val url = topicUrl(baseUrl, topicsStr)
@@ -65,7 +66,14 @@ class WsConnection(
         val nextId = ID.incrementAndGet()
         val sinceVal = if (since == 0L) "all" else since.toString()
         val urlWithSince = topicUrlWs(baseUrl, topicsStr, sinceVal)
-        val request = Request.Builder().url(urlWithSince).get().build()
+        val builder = Request.Builder()
+            .get()
+            .url(urlWithSince)
+            .addHeader("User-Agent", ApiService.USER_AGENT)
+        if (user != null) {
+            builder.addHeader("Authorization", Credentials.basic(user.username, user.password, StandardCharsets.UTF_8))
+        }
+        val request = builder.build()
         Log.d(TAG, "[$url] WebSocket($nextId): opening $urlWithSince ...")
         webSocket = client.newWebSocket(request, Listener(nextId))
     }
@@ -87,10 +95,6 @@ class WsConnection(
         return since
     }
 
-    override fun matches(otherSubscriptionIds: Collection<Long>): Boolean {
-        return subscriptionIds.toSet() == otherSubscriptionIds.toSet()
-    }
-
     @Synchronized
     fun scheduleReconnect(seconds: Int) {
         if (closed || state == State.Connecting || state == State.Connected) {

app/src/main/java/io/heckel/ntfy/ui/AddFragment.kt

@@ -15,13 +15,24 @@ import com.google.android.material.textfield.TextInputLayout
 import io.heckel.ntfy.BuildConfig
 import io.heckel.ntfy.R
 import io.heckel.ntfy.db.Repository
+import io.heckel.ntfy.db.User
+import io.heckel.ntfy.log.Log
+import io.heckel.ntfy.msg.ApiService
+import io.heckel.ntfy.util.topicUrl
+import kotlinx.android.synthetic.main.fragment_add_dialog.*
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
+import kotlin.random.Random
 
 class AddFragment : DialogFragment() {
+    private val api = ApiService()
+
     private lateinit var repository: Repository
     private lateinit var subscribeListener: SubscribeListener
 
+    private lateinit var subscribeView: View
+    private lateinit var loginView: View
+
     private lateinit var topicNameText: TextInputEditText
     private lateinit var baseUrlLayout: TextInputLayout
     private lateinit var baseUrlText: AutoCompleteTextView
@@ -32,10 +43,16 @@ class AddFragment : DialogFragment() {
     private lateinit var instantDeliveryDescription: View
     private lateinit var subscribeButton: Button
 
+    private lateinit var users: List<User>
+    private lateinit var usersSpinner: Spinner
+    private var userSelected: User? = null
+    private lateinit var usernameText: TextInputEditText
+    private lateinit var passwordText: TextInputEditText
+
     private lateinit var baseUrls: List<String> // List of base URLs already used, excluding app_base_url
 
     interface SubscribeListener {
-        fun onSubscribe(topic: String, baseUrl: String, instant: Boolean)
+        fun onSubscribe(topic: String, baseUrl: String, instant: Boolean, authUserId: Long?)
     }
 
     override fun onAttach(context: Context) {
@@ -53,6 +70,13 @@ class AddFragment : DialogFragment() {
 
         // Build root view
         val view = requireActivity().layoutInflater.inflate(R.layout.fragment_add_dialog, null)
+
+        // Main "pages"
+        subscribeView = view.findViewById(R.id.add_dialog_subscribe_view)
+        loginView = view.findViewById(R.id.add_dialog_login_view)
+        loginView.visibility = View.GONE
+
+        // Fields for "subscribe page"
         topicNameText = view.findViewById(R.id.add_dialog_topic_text)
         baseUrlLayout = view.findViewById(R.id.add_dialog_base_url_layout)
         baseUrlText = view.findViewById(R.id.add_dialog_base_url_text)
@@ -62,6 +86,11 @@ class AddFragment : DialogFragment() {
         useAnotherServerCheckbox = view.findViewById(R.id.add_dialog_use_another_server_checkbox)
         useAnotherServerDescription = view.findViewById(R.id.add_dialog_use_another_server_description)
 
+        // Fields for "login page"
+        usersSpinner = view.findViewById(R.id.add_dialog_login_users_spinner)
+        usernameText = view.findViewById(R.id.add_dialog_login_username)
+        passwordText = view.findViewById(R.id.add_dialog_login_password)
+
         // Set "Use another server" description based on flavor
         useAnotherServerDescription.text = if (BuildConfig.FIREBASE_AVAILABLE) {
             getString(R.string.add_dialog_use_another_server_description)
@@ -105,8 +134,9 @@ class AddFragment : DialogFragment() {
             }
         })
 
-        // Fill autocomplete for base URL
+        // Fill autocomplete for base URL & users drop-down
         lifecycleScope.launch(Dispatchers.IO) {
+            // Auto-complete
             val appBaseUrl = getString(R.string.app_base_url)
             baseUrls = repository.getSubscriptions()
                 .groupBy { it.baseUrl }
@@ -126,23 +156,46 @@ class AddFragment : DialogFragment() {
                     baseUrlLayout.setEndIconDrawable(0)
                 }
             }
+
+            // Users dropdown
+            users = repository.getUsers()
+            if (users.isEmpty()) {
+                usersSpinner.visibility = View.GONE
+            } else {
+                val spinnerEntries = users
+                    //.map { it.username }
+                    .toMutableList()
+                spinnerEntries.add(0, User(0, "Create new", ""))
+                usersSpinner.adapter = ArrayAdapter(requireActivity(), R.layout.fragment_add_dialog_dropdown_item, spinnerEntries)
+            }
         }
 
         // Show/hide based on flavor
         instantDeliveryBox.visibility = if (BuildConfig.FIREBASE_AVAILABLE) View.VISIBLE else View.GONE
 
+        // Show/hide spinner and username/password fields
+        usersSpinner.onItemSelectedListener = object : AdapterView.OnItemSelectedListener {
+            override fun onItemSelected(parent: AdapterView<*>?, view: View?, position: Int, id: Long) {
+                if (position == 0) {
+                    userSelected = null
+                    usernameText.visibility = View.VISIBLE
+                    passwordText.visibility = View.VISIBLE
+                } else {
+                    userSelected = usersSpinner.selectedItem as User
+                    usernameText.visibility = View.GONE
+                    passwordText.visibility = View.GONE
+                }
+            }
+            override fun onNothingSelected(parent: AdapterView<*>?) {
+                // This should not happen, ha!
+            }
+        }
+
         // Build dialog
         val alert = AlertDialog.Builder(activity)
             .setView(view)
             .setPositiveButton(R.string.add_dialog_button_subscribe) { _, _ ->
-                val topic = topicNameText.text.toString()
+                // This will be overridden below to avoid closing the dialog immediately
-                val baseUrl = getBaseUrl()
-                val instant = if (!BuildConfig.FIREBASE_AVAILABLE || useAnotherServerCheckbox.isChecked) {
-                    true
-                } else {
-                    instantDeliveryCheckbox.isChecked
-                }
-                subscribeListener.onSubscribe(topic, baseUrl, instant)
             }
             .setNegativeButton(R.string.add_dialog_button_cancel) { _, _ ->
                 dialog?.cancel()
@@ -155,6 +208,9 @@ class AddFragment : DialogFragment() {
 
             subscribeButton = dialog.getButton(AlertDialog.BUTTON_POSITIVE)
             subscribeButton.isEnabled = false
+            subscribeButton.setOnClickListener {
+                subscribeButtonClick()
+            }
 
             val textWatcher = object : TextWatcher {
                 override fun afterTextChanged(s: Editable?) {
@@ -193,6 +249,61 @@ class AddFragment : DialogFragment() {
         return alert
     }
 
+    private fun subscribeButtonClick() {
+        val topic = topicNameText.text.toString()
+        val baseUrl = getBaseUrl()
+        if (subscribeView.visibility == View.VISIBLE) {
+            checkAnonReadAndMaybeShowLogin(baseUrl, topic)
+        } else if (loginView.visibility == View.VISIBLE) {
+            checkAuthAndMaybeDismiss(baseUrl, topic)
+        }
+    }
+
+    private fun checkAnonReadAndMaybeShowLogin(baseUrl: String, topic: String) {
+        lifecycleScope.launch(Dispatchers.IO) {
+            Log.d(TAG, "Checking anonymous read access to topic ${topicUrl(baseUrl, topic)}")
+            val authorized = api.checkAnonTopicRead(baseUrl, topic)
+            if (authorized) {
+                Log.d(TAG, "Anonymous access granted to topic ${topicUrl(baseUrl, topic)}")
+                dismiss(authUserId = null)
+            } else {
+                Log.w(TAG, "Anonymous access not allowed to topic ${topicUrl(baseUrl, topic)}, showing login dialog")
+                requireActivity().runOnUiThread {
+                    subscribeView.visibility = View.GONE
+                    loginView.visibility = View.VISIBLE
+                }
+            }
+        }
+    }
+
+    private fun checkAuthAndMaybeDismiss(baseUrl: String, topic: String) {
+        val existingUser = usersSpinner.selectedItem != null && usersSpinner.selectedItem is User && usersSpinner.selectedItemPosition > 0
+        val user = if (existingUser) {
+            usersSpinner.selectedItem as User
+        } else {
+            User(
+                id = Random.nextLong(),
+                username = usernameText.text.toString(),
+                password = passwordText.text.toString()
+            )
+        }
+        lifecycleScope.launch(Dispatchers.IO) {
+            Log.d(TAG, "Checking read access for user ${user.username} to topic ${topicUrl(baseUrl, topic)}")
+            val authorized = api.checkUserTopicRead(baseUrl, topic, user.username, user.password)
+            if (authorized) {
+                Log.d(TAG, "Access granted for user ${user.username} to topic ${topicUrl(baseUrl, topic)}")
+                if (!existingUser) {
+                    Log.d(TAG, "Adding new user ${user.username} to database")
+                    repository.addUser(user)
+                }
+                dismiss(authUserId = user.id)
+            } else {
+                Log.w(TAG, "Access not allowed for user ${user.username} to topic ${topicUrl(baseUrl, topic)}")
+                // Show some error message
+            }
+        }
+    }
+
     private fun validateInput() = lifecycleScope.launch(Dispatchers.IO) {
         val baseUrl = getBaseUrl()
         val topic = topicNameText.text.toString()
@@ -215,6 +326,21 @@ class AddFragment : DialogFragment() {
         }
     }
 
+    private fun dismiss(authUserId: Long?) {
+        Log.d(TAG, "Closing dialog and calling onSubscribe handler")
+        requireActivity().runOnUiThread {
+            val topic = topicNameText.text.toString()
+            val baseUrl = getBaseUrl()
+            val instant = if (!BuildConfig.FIREBASE_AVAILABLE || useAnotherServerCheckbox.isChecked) {
+                true
+            } else {
+                instantDeliveryCheckbox.isChecked
+            }
+            subscribeListener.onSubscribe(topic, baseUrl, instant, authUserId = authUserId)
+            dialog?.dismiss()
+        }
+    }
+
     private fun getBaseUrl(): String {
         return if (useAnotherServerCheckbox.isChecked) {
             baseUrlText.text.toString()

app/src/main/java/io/heckel/ntfy/ui/MainActivity.kt

@@ -348,7 +348,7 @@ class MainActivity : AppCompatActivity(), ActionMode.Callback, AddFragment.Subsc
         newFragment.show(supportFragmentManager, AddFragment.TAG)
     }
 
-    override fun onSubscribe(topic: String, baseUrl: String, instant: Boolean) {
+    override fun onSubscribe(topic: String, baseUrl: String, instant: Boolean, authUserId: Long?) {
         Log.d(TAG, "Adding subscription ${topicShortUrl(baseUrl, topic)}")
 
         // Add subscription to database
@@ -358,6 +358,7 @@ class MainActivity : AppCompatActivity(), ActionMode.Callback, AddFragment.Subsc
             topic = topic,
             instant = instant,
             mutedUntil = 0,
+            authUserId = authUserId,
             upAppId = null,
             upConnectorToken = null,
             totalCount = 0,

app/src/main/java/io/heckel/ntfy/ui/NotificationFragment.kt

@@ -45,7 +45,7 @@ class NotificationFragment : DialogFragment() {
         // Dependencies
         val database = Database.getInstance(requireActivity().applicationContext)
         val sharedPrefs = requireActivity().getSharedPreferences(Repository.SHARED_PREFS_ID, Context.MODE_PRIVATE)
-        repository = Repository.getInstance(sharedPrefs, database.subscriptionDao(), database.notificationDao())
+        repository = Repository.getInstance(sharedPrefs, database)
 
         // Build root view
         val view = requireActivity().layoutInflater.inflate(R.layout.fragment_notification_dialog, null)

app/src/main/java/io/heckel/ntfy/up/BroadcastReceiver.kt

@@ -67,6 +67,7 @@ class BroadcastReceiver : android.content.BroadcastReceiver() {
                 topic = topic,
                 instant = true, // No Firebase, always instant!
                 mutedUntil = 0,
+                authUserId = null, // FIXME add UP user in settings
                 upAppId = appId,
                 upConnectorToken = connectorToken,
                 totalCount = 0,

app/src/main/java/io/heckel/ntfy/util/Util.kt

@@ -27,6 +27,7 @@ fun topicUrl(baseUrl: String, topic: String) = "${baseUrl}/${topic}"
 fun topicUrlUp(baseUrl: String, topic: String) = "${baseUrl}/${topic}?up=1" // UnifiedPush
 fun topicUrlJson(baseUrl: String, topic: String, since: String) = "${topicUrl(baseUrl, topic)}/json?since=$since"
 fun topicUrlWs(baseUrl: String, topic: String, since: String) = "${topicUrl(baseUrl, topic)}/ws?since=$since"
+fun topicUrlAuth(baseUrl: String, topic: String) = "${topicUrl(baseUrl, topic)}/auth"
 fun topicUrlJsonPoll(baseUrl: String, topic: String, since: String) = "${topicUrl(baseUrl, topic)}/json?poll=1&since=$since"
 fun topicShortUrl(baseUrl: String, topic: String) = shortUrl(topicUrl(baseUrl, topic))
 

app/src/main/java/io/heckel/ntfy/work/PollWorker.kt

@@ -28,7 +28,7 @@ class PollWorker(ctx: Context, params: WorkerParameters) : CoroutineWorker(ctx,
             Log.d(TAG, "Polling for new notifications")
             val database = Database.getInstance(applicationContext)
             val sharedPrefs = applicationContext.getSharedPreferences(Repository.SHARED_PREFS_ID, Context.MODE_PRIVATE)
-            val repository = Repository.getInstance(sharedPrefs, database.subscriptionDao(), database.notificationDao())
+            val repository = Repository.getInstance(sharedPrefs, database)
             val dispatcher = NotificationDispatcher(applicationContext, repository)
             val api = ApiService()
 

app/src/main/res/layout/fragment_add_dialog.xml

@@ -7,6 +7,10 @@
               android:paddingLeft="16dp"
               android:paddingRight="16dp">
 
+    <LinearLayout
+            android:orientation="vertical"
+            android:layout_width="match_parent"
+            android:layout_height="match_parent" android:id="@+id/add_dialog_subscribe_view" tools:visibility="gone">
         <TextView
                 android:id="@+id/add_dialog_title_text"
                 android:layout_width="match_parent"
@@ -77,7 +81,8 @@
                     android:text="@string/add_dialog_instant_delivery"
                     android:layout_width="wrap_content"
                     android:layout_height="wrap_content" android:id="@+id/add_dialog_instant_delivery_checkbox"
-                android:layout_marginTop="-8dp" android:layout_marginBottom="-5dp" android:layout_marginStart="-3dp"/>
+                    android:layout_marginTop="-8dp" android:layout_marginBottom="-5dp"
+                    android:layout_marginStart="-3dp"/>
             <ImageView
                     android:layout_width="24dp"
                     android:layout_height="24dp" app:srcCompat="@drawable/ic_bolt_gray_24dp"
@@ -92,4 +97,39 @@
                 android:layout_height="wrap_content" android:id="@+id/add_dialog_instant_delivery_description"
                 android:paddingStart="4dp" android:paddingTop="0dp" android:layout_marginTop="-5dp"
                 android:visibility="gone"/>
+    </LinearLayout>
+    <LinearLayout
+            android:orientation="vertical"
+            android:layout_width="match_parent"
+            android:layout_height="match_parent" android:id="@+id/add_dialog_login_view" tools:visibility="visible">
+        <TextView
+                android:id="@+id/add_dialog_login_title"
+                android:layout_width="match_parent"
+                android:layout_height="wrap_content"
+                android:paddingTop="16dp"
+                android:paddingBottom="3dp"
+                android:text="Login required"
+                android:textAlignment="viewStart"
+                android:textAppearance="@style/TextAppearance.AppCompat.Large" android:paddingStart="4dp"/>
+        <TextView
+                android:text="This topic requires you to login. Please pick an existing user or type in a username and password."
+                android:layout_width="match_parent"
+                android:layout_height="wrap_content" android:id="@+id/add_dialog_login_description"
+                android:paddingStart="4dp" android:paddingTop="3dp"/>
+        <Spinner
+                android:layout_width="match_parent"
+                android:layout_height="wrap_content" android:id="@+id/add_dialog_login_users_spinner"/>
+        <com.google.android.material.textfield.TextInputEditText
+                android:id="@+id/add_dialog_login_username"
+                android:layout_width="match_parent"
+                android:layout_height="wrap_content" android:hint="Username"
+                android:importantForAutofill="no"
+                android:maxLines="1" android:inputType="text" android:maxLength="64"/>
+        <com.google.android.material.textfield.TextInputEditText
+                android:id="@+id/add_dialog_login_password"
+                android:layout_width="match_parent"
+                android:layout_height="wrap_content" android:hint="Password"
+                android:importantForAutofill="no"
+                android:maxLines="1" android:inputType="textPassword"/>
+    </LinearLayout>
 </LinearLayout>

app/src/main/res/values/strings.xml

@@ -85,7 +85,7 @@
     <string name="add_dialog_use_another_server">Use another server</string>
     <string name="add_dialog_use_another_server_description">
         You can subscribe to topics from your own server. This option requires a foreground service and
-        consumes more power, but also delivers notifications faster (even in doze mode).
+        consumes more power, but also delivers notifications faster.
     </string>
     <string name="add_dialog_use_another_server_description_noinstant">
         You can subscribe to topics from your own server. Simply type in the base