Fitbit Subscription API

build.gradle.kts

@@ -57,14 +57,12 @@ dependencies {
     implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-yaml:$jacksonVersion")
 
     val grizzlyVersion: String by project
-    runtimeOnly("org.glassfish.grizzly:grizzly-framework-monitoring:$grizzlyVersion")
-    runtimeOnly("org.glassfish.grizzly:grizzly-http-monitoring:$grizzlyVersion")
-    runtimeOnly("org.glassfish.grizzly:grizzly-http-server-monitoring:$grizzlyVersion")
-
-    val log4j2Version: String by project
-    runtimeOnly("org.apache.logging.log4j:log4j-slf4j-impl:$log4j2Version")
-    runtimeOnly("org.apache.logging.log4j:log4j-api:$log4j2Version")
-    runtimeOnly("org.apache.logging.log4j:log4j-jul:$log4j2Version")
+    implementation("org.glassfish.grizzly:grizzly-framework-monitoring:$grizzlyVersion")
+    implementation("org.glassfish.grizzly:grizzly-http-monitoring:$grizzlyVersion")
+    implementation("org.glassfish.grizzly:grizzly-http-server-monitoring:$grizzlyVersion")
+
+    val logbackVersion: String by project
+    runtimeOnly("ch.qos.logback:logback-classic:$logbackVersion")
 
     val jedisVersion: String by project
     implementation("redis.clients:jedis:$jedisVersion")

gateway.yml

@@ -38,3 +38,12 @@ pushIntegration:
                 uri: redis://localhost:6379
                 # Key prefix for locks
                 lockPrefix: radar-push-garmin/lock/
+    fitbit:
+        enabled: true
+        verificationCode: ""
+        clientId: ""
+        clientSecret: ""
+        userRepositoryUrl: ""
+        userRepositoryClientId: ""
+        userRepositoryClientSecret: ""
+        userRepositoryTokenUrl: ""

gradle.properties

@@ -3,13 +3,13 @@ dockerComposeStopContainers=true
 
 kotlinVersion=1.6.10
 okhttp3Version=4.9.3
-radarJerseyVersion=0.8.1
+radarJerseyVersion=0.9.1
 radarCommonsVersion=0.13.2
-radarSchemasVersion=0.7.6
+radarSchemasVersion=0.8.1
 radarOauthClientVersion=0.8.0
-jacksonVersion=2.12.2
-slf4jVersion=1.7.32
-log4j2Version=2.17.0
+jacksonVersion=2.13.4
+slf4jVersion=2.0.3
+logbackVersion=1.4.4
 kafkaVersion=2.8.1
 confluentVersion=6.2.0
 junitVersion=5.7.2

src/main/kotlin/org/radarbase/gateway/Config.kt

@@ -5,8 +5,10 @@ import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig.SCHEMA_REGI
 import org.apache.kafka.clients.CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG
 import org.radarbase.gateway.inject.PushIntegrationEnhancerFactory
 import org.radarbase.jersey.enhancer.EnhancerFactory
+import org.radarbase.push.integration.fitbit.user.FitbitUserRepository
 import org.radarbase.push.integration.garmin.user.GarminUserRepository
 import java.net.URI
+import java.time.Duration
 import java.time.Instant
 
 data class Config(
@@ -33,10 +35,12 @@ data class Config(
 }
 
 data class PushIntegrationConfig(
-    val garmin: GarminConfig = GarminConfig()
+    val garmin: GarminConfig = GarminConfig(),
+    val fitbit: FitbitConfig = FitbitConfig()
 ) {
     fun validate() {
         garmin.validate()
+        fitbit.validate()
         // Add more validations as services are added
     }
 }
@@ -81,6 +85,48 @@ data class GarminConfig(
     }
 }
 
+data class FitbitConfig(
+    val enabled: Boolean = false,
+    val verificationCode: String = "",
+    val clientSecret: String = "",
+    val clientId: String = "",
+    val subscriptionConfig: SubscriptionConfig = SubscriptionConfig(),
+    val userRepositoryClass: String =
+        "org.radarbase.push.integration.fitbit.user.FitbitUserRepository",
+    val userRepositoryUrl: String = "http://localhost:8080/",
+    val userRepositoryClientId: String = "radar_pushendpoint",
+    val userRepositoryClientSecret: String = "",
+    val userRepositoryTokenUrl: String = "http://localhost:8080/token/",
+    val sleepStagesTopic: String = "connect_fitbit_sleep_stages",
+    val sleepClassicTopic: String = "connect_fitbit_sleep_classic",
+    val activityLogTopic: String = "connect_fitbit_activity_log",
+    val foodLogTopic: String = "connect_fitbit_food_log",
+    val routePollIntervalMs: Long = 5000,
+    val pollIntervalPerUserSeconds: Long = 150,
+    val requestMaxThreads: Int = 4,
+    val redis: RedisConfig = RedisConfig(lockPrefix = "radar-fitbit-subscription/lock"),
+    val baseUrl: String = "https://api.fitbit.com",
+) {
+    val userRepository: Class<*> = Class.forName(userRepositoryClass)
+    val routePollInterval: Duration = Duration.ofMillis(routePollIntervalMs)
+    val pollIntervalPerUser: Duration = Duration.ofSeconds(pollIntervalPerUserSeconds)
+    val tooManyRequestsCooldown: Duration = Duration.ofHours(1)
+
+    fun validate() {
+        if (enabled) {
+            check(FitbitUserRepository::class.java.isAssignableFrom(userRepository)) {
+                "$userRepositoryClass is not valid. Please specify a class that is a subclass of" +
+                        " `org.radarbase.push.integration.fitbit.user.FitbitUserRepository`"
+            }
+        }
+    }
+}
+
+data class SubscriptionConfig(
+    val maxThreads: Int = 4,
+    val subscriberID: String = "1",
+)
+
 data class BackfillConfig(
     val enabled: Boolean = false,
     val redis: RedisConfig = RedisConfig(),

src/main/kotlin/org/radarbase/gateway/inject/PushIntegrationEnhancerFactory.kt

@@ -2,10 +2,10 @@ package org.radarbase.gateway.inject
 
 import okhttp3.internal.toImmutableList
 import org.radarbase.gateway.Config
-import org.radarbase.jersey.config.ConfigLoader
 import org.radarbase.jersey.enhancer.Enhancers
 import org.radarbase.jersey.enhancer.EnhancerFactory
 import org.radarbase.jersey.enhancer.JerseyResourceEnhancer
+import org.radarbase.push.integration.FitbitPushIntegrationResourceEnhancer
 import org.radarbase.push.integration.GarminPushIntegrationResourceEnhancer
 import org.radarbase.push.integration.common.inject.PushIntegrationResourceEnhancer
 
@@ -24,6 +24,9 @@ class PushIntegrationEnhancerFactory(private val config: Config) : EnhancerFacto
         if (config.pushIntegration.garmin.enabled) {
             enhancersList.add(GarminPushIntegrationResourceEnhancer(config))
         }
+        if (config.pushIntegration.fitbit.enabled) {
+            enhancersList.add(FitbitPushIntegrationResourceEnhancer(config))
+        }
         // Add more enhancers as services are added
 
         return enhancersList.toImmutableList()

src/main/kotlin/org/radarbase/push/integration/FitbitPushIntegrationResourceEnhancer.kt

@@ -0,0 +1,54 @@
+package org.radarbase.push.integration
+
+import com.fasterxml.jackson.databind.JsonNode
+import jakarta.inject.Singleton
+import org.glassfish.hk2.api.TypeLiteral
+import org.glassfish.jersey.internal.inject.AbstractBinder
+import org.glassfish.jersey.process.internal.RequestScoped
+import org.glassfish.jersey.server.ResourceConfig
+import org.radarbase.gateway.Config
+import org.radarbase.jersey.auth.AuthValidator
+import org.radarbase.jersey.enhancer.JerseyResourceEnhancer
+import org.radarbase.push.integration.common.auth.DelegatedAuthValidator.Companion.FITBIT_QUALIFIER
+import org.radarbase.push.integration.common.user.User
+import org.radarbase.push.integration.fitbit.auth.FitbitAuthValidator
+import org.radarbase.push.integration.fitbit.factory.FitbitUserTreeMapFactory
+import org.radarbase.push.integration.fitbit.service.fitbitapi.FitbitApiService
+import org.radarbase.push.integration.fitbit.service.fitbitapi.FitbitRequestProcessor
+import org.radarbase.push.integration.fitbit.user.FitbitUserRepository
+
+class FitbitPushIntegrationResourceEnhancer(private val config: Config) : JerseyResourceEnhancer {
+    override fun ResourceConfig.enhance() {
+        packages(
+            "org.radarbase.push.integration.fitbit.resource",
+            "org.radarbase.push.integration.common.filter",
+            "org.radarbase.push.integration.fitbit.filter"
+        )
+    }
+
+    override val classes: Array<Class<*>>
+        get() = arrayOf(FitbitRequestProcessor::class.java)
+
+    override fun AbstractBinder.enhance() {
+
+        bind(config.pushIntegration.fitbit.userRepository)
+            .to(FitbitUserRepository::class.java)
+            .named(FITBIT_QUALIFIER)
+            .`in`(Singleton::class.java)
+
+        bind(FitbitAuthValidator::class.java)
+            .to(AuthValidator::class.java)
+            .named(FITBIT_QUALIFIER)
+            .`in`(Singleton::class.java)
+
+        bind(FitbitApiService::class.java)
+            .to(FitbitApiService::class.java)
+            .`in`(Singleton::class.java)
+
+        bindFactory(FitbitUserTreeMapFactory::class.java)
+            .to(object : TypeLiteral<MutableMap<User, JsonNode>>() {}.type)
+            .proxy(true)
+            .named(FITBIT_QUALIFIER)
+            .`in`(RequestScoped::class.java)
+    }
+}

src/main/kotlin/org/radarbase/push/integration/common/auth/DelegatedAuthValidator.kt

@@ -17,6 +17,7 @@ class DelegatedAuthValidator(
     fun delegate(): AuthValidator {
         return when {
             uriInfo.matches(GARMIN_QUALIFIER) -> namedValidators.named(GARMIN_QUALIFIER).get()
+            uriInfo.matches(FITBIT_QUALIFIER) -> namedValidators.named(FITBIT_QUALIFIER).get()
             // Add support for more as integrations are added
             else -> throw IllegalStateException()
         }
@@ -27,6 +28,7 @@ class DelegatedAuthValidator(
 
     companion object {
         const val GARMIN_QUALIFIER = "garmin"
+        const val FITBIT_QUALIFIER = "fitbit"
     }
 
     override fun verify(token: String, request: ContainerRequestContext): Auth? =

src/main/kotlin/org/radarbase/push/integration/common/filter/CorsFilter.kt

@@ -16,7 +16,7 @@ class CorsFilter : ContainerResponseFilter {
         cres.headers
             .add("Access-Control-Allow-Headers", "origin, content-type, accept, authorization")
         cres.headers.add("Access-Control-Allow-Credentials", "true")
-        cres.headers.add("Access-Control-Allow-Methods", "POST, OPTIONS")
+        cres.headers.add("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
         cres.headers.add("Access-Control-Max-Age", "1209600")
     }
 }

src/main/kotlin/org/radarbase/push/integration/garmin/util/RedisHolder.kt → src/main/kotlin/org/radarbase/push/integration/common/redis/RedisHolder.kt

@@ -1,4 +1,4 @@
-package org.radarbase.push.integration.garmin.util
+package org.radarbase.push.integration.common.redis
 
 import redis.clients.jedis.Jedis
 import redis.clients.jedis.JedisPool

src/main/kotlin/org/radarbase/push/integration/garmin/util/RedisRemoteLockManager.kt → src/main/kotlin/org/radarbase/push/integration/common/redis/RedisRemoteLockManager.kt

@@ -1,13 +1,13 @@
-package org.radarbase.push.integration.garmin.util
+package org.radarbase.push.integration.common.redis
 
 import org.slf4j.LoggerFactory
 import redis.clients.jedis.params.SetParams
 import java.time.Duration
 import java.util.*
 
 class RedisRemoteLockManager(
-        private val redisHolder: RedisHolder,
-        private val keyPrefix: String
+    private val redisHolder: RedisHolder,
+    private val keyPrefix: String
 ) : RemoteLockManager {
     private val uuid: String = UUID.randomUUID().toString()
 

src/main/kotlin/org/radarbase/push/integration/garmin/util/RemoteLockManager.kt → src/main/kotlin/org/radarbase/push/integration/common/redis/RemoteLockManager.kt

@@ -1,4 +1,4 @@
-package org.radarbase.push.integration.garmin.util
+package org.radarbase.push.integration.common.redis
 
 import java.io.Closeable
 

src/main/kotlin/org/radarbase/push/integration/fitbit/auth/FitbitAuthValidator.kt

@@ -0,0 +1,119 @@
+package org.radarbase.push.integration.fitbit.auth
+
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.ObjectMapper
+import jakarta.inject.Named
+import jakarta.ws.rs.container.ContainerRequestContext
+import jakarta.ws.rs.core.Context
+import org.radarbase.gateway.Config
+import org.radarbase.jersey.auth.Auth
+import org.radarbase.jersey.auth.AuthValidator
+import org.radarbase.jersey.auth.disabled.DisabledAuth
+import org.radarbase.jersey.exception.HttpNotFoundException
+import org.radarbase.push.integration.common.auth.DelegatedAuthValidator.Companion.FITBIT_QUALIFIER
+import org.radarbase.push.integration.common.user.User
+import org.radarbase.push.integration.fitbit.user.FitbitUserRepository
+import java.security.InvalidKeyException
+import java.security.NoSuchAlgorithmException
+import java.util.Base64
+import javax.crypto.Mac
+import javax.crypto.spec.SecretKeySpec
+
+class FitbitAuthValidator(
+    @Context val objectMapper: ObjectMapper,
+    @Context val config: Config,
+    @Named(FITBIT_QUALIFIER) private val userRepository: FitbitUserRepository
+) : AuthValidator {
+
+    override fun verify(token: String, request: ContainerRequestContext): Auth {
+        return if (token.isBlank()) {
+            throw HttpNotFoundException("not_found", "Signature was not found")
+        } else {
+
+            val tree: JsonNode? = if (request.hasEntity()) {
+                // We put the json tree in the request because the entity stream will be closed here
+                val tree1 = objectMapper.readTree(request.entityStream)
+                request.setProperty("tree", tree1)
+                tree1
+            } else null
+
+            if (!isSignatureValid(token, tree)) {
+                throw HttpNotFoundException("invalid_signature", "Valid Signature not found")
+            }
+
+            if (!checkIsUserAuthorized(request, tree)) {
+                request.setProperty("user_tree_map", null)
+            }
+
+            // Disable auth since we don't have proper auth support
+            DisabledAuth("res_gateway")
+        }
+    }
+
+    override fun getToken(request: ContainerRequestContext): String =
+        request.getHeaderString("X-Fitbit-Signature")
+            ?: throw HttpNotFoundException("not_found", "Signature was not found")
+
+
+    fun checkIsUserAuthorized(request: ContainerRequestContext, tree: JsonNode?): Boolean {
+
+        if (tree == null) {
+            return false
+        }
+
+        val userTreeMap: Map<User, JsonNode> =
+            tree[tree.fieldNames().next()]
+                .groupBy { node ->
+                    node[USER_ID_KEY].asText()
+                }
+                .filter { (userId, userData) ->
+                    try {
+                        userRepository.findByExternalId(userId)
+                        true
+                    } catch (ex: NoSuchElementException) {
+                        false
+                    }
+                }
+                .entries
+                .associate { (userId, userData) ->
+                    userRepository.findByExternalId(userId) to
+                            objectMapper.createObjectNode()
+                                .set(tree.fieldNames().next(), objectMapper.valueToTree(userData))
+
+                }
+        request.setProperty("user_tree_map", userTreeMap)
+        return true
+    }
+
+
+    fun isSignatureValid(signature: String?, contents: JsonNode?): Boolean {
+        val signingKey = "${config.pushIntegration.fitbit.clientSecret}&"
+        if (signature == null) {
+            return false
+        }
+        if (contents == null) {
+            return false
+        }
+        val genHMAC = genHMAC(contents.asText(), signingKey)
+        return genHMAC.equals(signature)
+    }
+
+    fun genHMAC(data: String, key: String): String? {
+        var result: ByteArray? = null
+        try {
+            val signinKey = SecretKeySpec(key.toByteArray(), "HmacSHA1")
+            val mac = Mac.getInstance("HmacSHA1")
+            mac.init(signinKey)
+            val rawHmac = mac.doFinal(data.toByteArray())
+            result = Base64.getEncoder().encode(rawHmac)
+        } catch (e: NoSuchAlgorithmException) {
+            System.err.println(e.message)
+        } catch (e: InvalidKeyException) {
+            System.err.println(e.message)
+        }
+        return result?.let { String(it) }
+    }
+    companion object {
+        const val USER_ID_KEY = "ownerId"
+    }
+}