Add optional OpenTelemetry Netty server instrumentation

- Update lys.version to 2026.04.14

- Add optional compileOnly dependency on opentelemetry-netty-4.1 in rbcs-server

- Add runtime guard to only activate instrumentation when OTel classes are on classpath

- Insert OTel combined handler after HttpServerCodec in the Netty pipeline

- Add requires-static JPMS directives for optional module support

- Add enableTelemetry config attribute to rbcs:server with default false

- Update Configuration DTO, XSD schema, Parser, Serializer, and all tests
This commit is contained in:
OpenCode
2026-04-28 14:59:08 +00:00
parent 70eccf83a8
commit ee7bc7e850
198 changed files with 11040 additions and 4 deletions
@@ -0,0 +1,2 @@
net.woggioni.rbcs.server.cache.FileSystemCacheProvider
net.woggioni.rbcs.server.cache.InMemoryCacheProvider
@@ -0,0 +1,562 @@
package net.woggioni.rbcs.server
import io.netty.bootstrap.ServerBootstrap
import io.netty.buffer.ByteBuf
import io.netty.channel.Channel
import io.netty.channel.ChannelFactory
import io.netty.channel.ChannelFuture
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter
import io.netty.channel.ChannelInitializer
import io.netty.channel.ChannelOption
import io.netty.channel.ChannelPromise
import io.netty.channel.MultiThreadIoEventLoopGroup
import io.netty.channel.nio.NioIoHandler
import io.netty.channel.socket.DatagramChannel
import io.netty.channel.socket.ServerSocketChannel
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.NioDatagramChannel
import io.netty.channel.socket.nio.NioServerSocketChannel
import io.netty.channel.socket.nio.NioSocketChannel
import io.netty.handler.codec.compression.CompressionOptions
import io.netty.handler.codec.haproxy.HAProxyMessageDecoder
import io.netty.handler.codec.http.DefaultHttpContent
import io.netty.handler.codec.http.HttpContentCompressor
import io.netty.handler.codec.http.HttpDecoderConfig
import io.netty.handler.codec.http.HttpHeaderNames
import io.netty.handler.codec.http.HttpRequest
import io.netty.handler.codec.http.HttpServerCodec
import io.netty.handler.ssl.ClientAuth
import io.netty.handler.ssl.SslContext
import io.netty.handler.ssl.SslContextBuilder
import io.netty.handler.ssl.SslHandler
import io.netty.handler.stream.ChunkedWriteHandler
import io.netty.handler.timeout.IdleState
import io.netty.handler.timeout.IdleStateEvent
import io.netty.handler.timeout.IdleStateHandler
import io.netty.util.AttributeKey
import io.netty.util.concurrent.EventExecutorGroup
import java.io.OutputStream
import java.net.InetSocketAddress
import java.nio.file.Files
import java.nio.file.Path
import java.security.PrivateKey
import java.security.cert.X509Certificate
import java.time.Duration
import java.time.Instant
import java.util.Arrays
import java.util.Base64
import java.util.concurrent.CompletableFuture
import java.util.concurrent.Future
import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException
import java.util.regex.Matcher
import java.util.regex.Pattern
import javax.naming.ldap.LdapName
import javax.net.ssl.SSLPeerUnverifiedException
import net.woggioni.rbcs.api.AsyncCloseable
import net.woggioni.rbcs.api.Configuration
import net.woggioni.rbcs.api.exception.ConfigurationException
import net.woggioni.rbcs.common.Cidr
import net.woggioni.rbcs.common.PasswordSecurity.decodePasswordHash
import net.woggioni.rbcs.common.PasswordSecurity.hashPassword
import net.woggioni.rbcs.common.RBCS.getTrustManager
import net.woggioni.rbcs.common.RBCS.loadKeystore
import net.woggioni.rbcs.common.RBCS.toUrl
import net.woggioni.rbcs.common.Xml
import net.woggioni.rbcs.common.createLogger
import net.woggioni.rbcs.common.debug
import net.woggioni.rbcs.common.info
import net.woggioni.rbcs.server.otel.OtelIntegration
import net.woggioni.rbcs.server.auth.AbstractNettyHttpAuthenticator
import net.woggioni.rbcs.server.auth.Authorizer
import net.woggioni.rbcs.server.auth.RoleAuthorizer
import net.woggioni.rbcs.server.configuration.Parser
import net.woggioni.rbcs.server.configuration.Serializer
import net.woggioni.rbcs.server.exception.ExceptionHandler
import net.woggioni.rbcs.server.handler.MaxRequestSizeHandler
import net.woggioni.rbcs.server.handler.ProxyProtocolHandler
import net.woggioni.rbcs.server.handler.ReadTriggerDuplexHandler
import net.woggioni.rbcs.server.handler.ServerHandler
import net.woggioni.rbcs.server.throttling.BucketManager
import net.woggioni.rbcs.server.throttling.ThrottlingHandler
class RemoteBuildCacheServer(private val cfg: Configuration) {
companion object {
private val log = createLogger<RemoteBuildCacheServer>()
val userAttribute: AttributeKey<Configuration.User> = AttributeKey.valueOf("user")
val groupAttribute: AttributeKey<Set<Configuration.Group>> = AttributeKey.valueOf("group")
val clientIp: AttributeKey<InetSocketAddress> = AttributeKey.valueOf("client-ip")
val DEFAULT_CONFIGURATION_URL by lazy { "jpms://net.woggioni.rbcs.server/net/woggioni/rbcs/server/rbcs-default.xml".toUrl() }
private const val SSL_HANDLER_NAME = "sslHandler"
fun loadConfiguration(configurationFile: Path): Configuration {
val doc = Files.newInputStream(configurationFile).use {
Xml.parseXml(configurationFile.toUri().toURL(), it)
}
return Parser.parse(doc)
}
fun dumpConfiguration(conf: Configuration, outputStream: OutputStream) {
Xml.write(Serializer.serialize(conf), outputStream)
}
}
private class HttpChunkContentCompressor(
threshold: Int,
vararg compressionOptions: CompressionOptions = emptyArray()
) : HttpContentCompressor(threshold, *compressionOptions) {
override fun write(ctx: ChannelHandlerContext, msg: Any, promise: ChannelPromise) {
var message: Any? = msg
if (message is ByteBuf) {
// convert ByteBuf to HttpContent to make it work with compression. This is needed as we use the
// ChunkedWriteHandler to send files when compression is enabled.
val buff = message
if (buff.isReadable) {
// We only encode non empty buffers, as empty buffers can be used for determining when
// the content has been flushed and it confuses the HttpContentCompressor
// if we let it go
message = DefaultHttpContent(buff)
}
}
super.write(ctx, message, promise)
}
}
@Sharable
private class ClientCertificateAuthenticator(
authorizer: Authorizer,
private val anonymousUserGroups: Set<Configuration.Group>?,
private val userExtractor: Configuration.UserExtractor?,
private val groupExtractor: Configuration.GroupExtractor?,
) : AbstractNettyHttpAuthenticator(authorizer) {
override fun authenticate(ctx: ChannelHandlerContext, req: HttpRequest): AuthenticationResult? {
return try {
val sslHandler = (ctx.pipeline().get(SSL_HANDLER_NAME) as? SslHandler)
?: throw ConfigurationException("Client certificate authentication cannot be used when TLS is disabled")
val sslEngine = sslHandler.engine()
sslEngine.session.peerCertificates.takeIf {
it.isNotEmpty()
}?.let { peerCertificates ->
val clientCertificate = peerCertificates.first() as X509Certificate
val user = userExtractor?.extract(clientCertificate)
val group = groupExtractor?.extract(clientCertificate)
val allGroups =
((user?.groups ?: emptySet()).asSequence() + sequenceOf(group).filterNotNull()).toSet()
AuthenticationResult(user, allGroups)
} ?: anonymousUserGroups?.let { AuthenticationResult(null, it) }
} catch (ex: SSLPeerUnverifiedException) {
log.debug(ctx) {
ex.message ?: "Error witch client certificate authentication"
}
anonymousUserGroups?.let { AuthenticationResult(null, it) }
}
}
}
@Sharable
private class ForwardedClientCertificateAuthenticator(
authorizer: Authorizer,
private val anonymousUserGroups: Set<Configuration.Group>?,
private val subjectDnUserExtractor: SubjectDnExtractor?,
private val subjectDnGroupExtractor: SubjectDnExtractor?,
private val headerName: String,
private val trustedProxyIPs: List<Cidr>,
private val users: Map<String, Configuration.User>,
private val groups: Map<String, Configuration.Group>,
) : AbstractNettyHttpAuthenticator(authorizer) {
companion object {
private val log = createLogger<ForwardedClientCertificateAuthenticator>()
}
override fun authenticate(ctx: ChannelHandlerContext, req: HttpRequest): AuthenticationResult? {
val clientIp = ctx.channel().attr(clientIp).get()
if (clientIp == null || trustedProxyIPs.none { it.contains(clientIp.address) }) {
log.debug(ctx) {
"Rejecting forwarded client certificate authentication from untrusted address: $clientIp"
}
return null
}
val subjectDn = req.headers()[headerName]
?: return anonymousUserGroups?.let { AuthenticationResult(null, it) }
val ldapName = try {
LdapName(subjectDn)
} catch (e: Exception) {
log.debug(ctx) {
"Invalid subject DN in header $headerName: $subjectDn"
}
return anonymousUserGroups?.let { AuthenticationResult(null, it) }
}
val user = subjectDnUserExtractor?.extract(ldapName)?.let { userName ->
users[userName] ?: throw RuntimeException("Failed to extract user '$userName'")
}
val group = subjectDnGroupExtractor?.extract(ldapName)?.let { groupName ->
groups[groupName] ?: throw RuntimeException("Failed to extract group '$groupName'")
}
val allGroups = ((user?.groups ?: emptySet()).asSequence() + sequenceOf(group).filterNotNull()).toSet()
return AuthenticationResult(user, allGroups)
}
}
private data class SubjectDnExtractor(val rdnType: String, val pattern: Pattern) {
fun extract(ldapName: LdapName): String? {
return ldapName.rdns.find { it.type == rdnType }
?.let { pattern.matcher(it.value.toString()) }
?.takeIf(Matcher::matches)?.group(1)
}
}
@Sharable
private class NettyHttpBasicAuthenticator(
private val users: Map<String, Configuration.User>, authorizer: Authorizer
) : AbstractNettyHttpAuthenticator(authorizer) {
companion object {
private val log = createLogger<NettyHttpBasicAuthenticator>()
}
override fun authenticate(ctx: ChannelHandlerContext, req: HttpRequest): AuthenticationResult? {
val authorizationHeader = req.headers()[HttpHeaderNames.AUTHORIZATION] ?: let {
log.debug(ctx) {
"Missing Authorization header"
}
return users[""]?.let { AuthenticationResult(it, it.groups) }
}
val cursor = authorizationHeader.indexOf(' ')
if (cursor < 0) {
log.debug(ctx) {
"Invalid Authorization header: '$authorizationHeader'"
}
return users[""]?.let { AuthenticationResult(it, it.groups) }
}
val authenticationType = authorizationHeader.substring(0, cursor)
if ("Basic" != authenticationType) {
log.debug(ctx) {
"Invalid authentication type header: '$authenticationType'"
}
return users[""]?.let { AuthenticationResult(it, it.groups) }
}
val (username, password) = Base64.getDecoder().decode(authorizationHeader.substring(cursor + 1))
.let(::String)
.let {
val colon = it.indexOf(':')
if (colon < 0) {
log.debug(ctx) {
"Missing colon from authentication"
}
return null
}
it.substring(0, colon) to it.substring(colon + 1)
}
return username.let(users::get)?.takeIf { user ->
user.password?.let { passwordAndSalt ->
val (_, salt) = decodePasswordHash(passwordAndSalt)
hashPassword(password, Base64.getEncoder().encodeToString(salt)) == passwordAndSalt
} ?: false
}?.let { user ->
AuthenticationResult(user, user.groups)
}
}
}
private class ServerInitializer(
private val cfg: Configuration,
private val channelFactory : ChannelFactory<SocketChannel>,
private val datagramChannelFactory : ChannelFactory<DatagramChannel>,
) : ChannelInitializer<Channel>(), AsyncCloseable {
companion object {
private fun createSslCtx(tls: Configuration.Tls): SslContext {
val keyStore = tls.keyStore
return if (keyStore == null) {
throw IllegalArgumentException("No keystore configured")
} else {
val javaKeyStore = loadKeystore(keyStore.file, keyStore.password)
val serverKey = javaKeyStore.getKey(
keyStore.keyAlias, (keyStore.keyPassword ?: "").let(String::toCharArray)
) as PrivateKey
val serverCert: Array<X509Certificate> =
Arrays.stream(javaKeyStore.getCertificateChain(keyStore.keyAlias))
.map { it as X509Certificate }
.toArray { size -> Array<X509Certificate?>(size) { null } }
SslContextBuilder.forServer(serverKey, *serverCert).apply {
val clientAuth = tls.trustStore?.let { trustStore ->
val ts = loadKeystore(trustStore.file, trustStore.password)
trustManager(
getTrustManager(ts, trustStore.isCheckCertificateStatus)
)
if (trustStore.isRequireClientCertificate) ClientAuth.REQUIRE
else ClientAuth.OPTIONAL
} ?: ClientAuth.NONE
clientAuth(clientAuth)
}.build()
}
}
private val log = createLogger<ServerInitializer>()
}
private val cacheHandlerFactory = cfg.cache.materialize()
private val bucketManager = BucketManager.from(cfg)
private val authenticator = when (val auth = cfg.authentication) {
is Configuration.BasicAuthentication -> NettyHttpBasicAuthenticator(cfg.users, RoleAuthorizer())
is Configuration.ClientCertificateAuthentication -> {
ClientCertificateAuthenticator(
RoleAuthorizer(),
cfg.users[""]?.groups,
userExtractor(auth),
groupExtractor(auth)
)
}
is Configuration.ForwardedClientCertificateAuthentication -> {
ForwardedClientCertificateAuthenticator(
RoleAuthorizer(),
cfg.users[""]?.groups,
auth.userExtractor?.let { extractor ->
SubjectDnExtractor(extractor.rdnType, Pattern.compile(extractor.pattern))
},
auth.groupExtractor?.let { extractor ->
SubjectDnExtractor(extractor.rdnType, Pattern.compile(extractor.pattern))
},
auth.headerName,
cfg.trustedProxyIPs,
cfg.users,
cfg.groups,
)
}
else -> null
}
private val proxyProtocolEnabled: Boolean = cfg.isProxyProtocolEnabled
private val trustedProxyIPs: List<Cidr> = cfg.trustedProxyIPs
private val sslContext: SslContext? = cfg.tls?.let(Companion::createSslCtx)
private fun userExtractor(authentication: Configuration.ClientCertificateAuthentication) =
authentication.userExtractor?.let { extractor ->
val pattern = Pattern.compile(extractor.pattern)
val rdnType = extractor.rdnType
Configuration.UserExtractor { cert: X509Certificate ->
val userName = LdapName(cert.subjectX500Principal.name).rdns.find {
it.type == rdnType
}?.let {
pattern.matcher(it.value.toString())
}?.takeIf(Matcher::matches)?.group(1)
cfg.users[userName] ?: throw java.lang.RuntimeException("Failed to extract user")
}
}
private fun groupExtractor(authentication: Configuration.ClientCertificateAuthentication) =
authentication.groupExtractor?.let { extractor ->
val pattern = Pattern.compile(extractor.pattern)
val rdnType = extractor.rdnType
Configuration.GroupExtractor { cert: X509Certificate ->
val groupName = LdapName(cert.subjectX500Principal.name).rdns.find {
it.type == rdnType
}?.let {
pattern.matcher(it.value.toString())
}?.takeIf(Matcher::matches)?.group(1)
cfg.groups[groupName] ?: throw java.lang.RuntimeException("Failed to extract group")
}
}
override fun initChannel(ch: Channel) {
ch.attr(clientIp).set(ch.remoteAddress() as InetSocketAddress)
log.debug {
"Created connection ${ch.id().asShortText()} with ${ch.remoteAddress()}"
}
ch.closeFuture().addListener {
log.debug {
"Closed connection ${ch.id().asShortText()} with ${ch.remoteAddress()}"
}
}
ch.config().isAutoRead = false
val pipeline = ch.pipeline()
cfg.connection.also { conn ->
val readIdleTimeout = conn.readIdleTimeout.toMillis()
val writeIdleTimeout = conn.writeIdleTimeout.toMillis()
val idleTimeout = conn.idleTimeout.toMillis()
if (readIdleTimeout > 0 || writeIdleTimeout > 0 || idleTimeout > 0) {
pipeline.addLast(
IdleStateHandler(
true,
readIdleTimeout,
writeIdleTimeout,
idleTimeout,
TimeUnit.MILLISECONDS
)
)
}
}
pipeline.addLast(object : ChannelInboundHandlerAdapter() {
override fun userEventTriggered(ctx: ChannelHandlerContext, evt: Any) {
if (evt is IdleStateEvent) {
when (evt.state()) {
IdleState.READER_IDLE -> log.debug {
"Read timeout reached on channel ${ch.id().asShortText()}, closing the connection"
}
IdleState.WRITER_IDLE -> log.debug {
"Write timeout reached on channel ${ch.id().asShortText()}, closing the connection"
}
IdleState.ALL_IDLE -> log.debug {
"Idle timeout reached on channel ${ch.id().asShortText()}, closing the connection"
}
null -> throw IllegalStateException("This should never happen")
}
ctx.close()
}
}
})
if(proxyProtocolEnabled) {
pipeline.addLast(HAProxyMessageDecoder())
pipeline.addLast(ProxyProtocolHandler(trustedProxyIPs))
}
sslContext?.newHandler(ch.alloc())?.also {
pipeline.addLast(SSL_HANDLER_NAME, it)
}
val httpDecoderConfig = HttpDecoderConfig().apply {
maxChunkSize = cfg.connection.chunkSize
}
pipeline.addLast(HttpServerCodec(httpDecoderConfig))
OtelIntegration.createHandler()?.let { pipeline.addLast(it) }
pipeline.addLast(ReadTriggerDuplexHandler.NAME, ReadTriggerDuplexHandler())
pipeline.addLast(MaxRequestSizeHandler.NAME, MaxRequestSizeHandler(cfg.connection.maxRequestSize))
pipeline.addLast(HttpChunkContentCompressor(1024))
pipeline.addLast(ChunkedWriteHandler())
authenticator?.let {
pipeline.addLast(it)
}
pipeline.addLast(ThrottlingHandler(bucketManager,cfg.rateLimiter, cfg.connection))
val serverHandler = let {
val prefix = Path.of("/").resolve(Path.of(cfg.serverPath ?: "/"))
ServerHandler(prefix) {
cacheHandlerFactory.newHandler(cfg, ch.eventLoop(), channelFactory, datagramChannelFactory)
}
}
pipeline.addLast(ServerHandler.NAME, serverHandler)
pipeline.addLast(ExceptionHandler.NAME, ExceptionHandler)
}
override fun asyncClose() = cacheHandlerFactory.asyncClose()
}
class ServerHandle(
closeFuture: ChannelFuture,
private val bossGroup: EventExecutorGroup,
private val executorGroups: Iterable<EventExecutorGroup>,
private val serverInitializer: AsyncCloseable,
) : Future<Void> by from(closeFuture, bossGroup, executorGroups, serverInitializer) {
companion object {
private val log = createLogger<ServerHandle>()
private fun from(
closeFuture: ChannelFuture,
bossGroup: EventExecutorGroup,
executorGroups: Iterable<EventExecutorGroup>,
serverInitializer: AsyncCloseable
): CompletableFuture<Void> {
val result = CompletableFuture<Void>()
closeFuture.addListener {
val errors = mutableListOf<Throwable>()
val deadline = Instant.now().plusSeconds(20)
serverInitializer.asyncClose().whenCompleteAsync { _, ex ->
if(ex != null) {
log.error(ex.message, ex)
errors.addLast(ex)
}
executorGroups.forEach(EventExecutorGroup::shutdownGracefully)
bossGroup.terminationFuture().sync()
for (executorGroup in executorGroups) {
val future = executorGroup.terminationFuture()
try {
val now = Instant.now()
if (now > deadline) {
future.get(0, TimeUnit.SECONDS)
} else {
future.get(Duration.between(now, deadline).toMillis(), TimeUnit.MILLISECONDS)
}
}
catch (te: TimeoutException) {
errors.addLast(te)
log.warn("Timeout while waiting for shutdown of $executorGroup", te)
} catch (ex: Throwable) {
log.warn(ex.message, ex)
errors.addLast(ex)
}
}
if(errors.isEmpty()) {
result.complete(null)
} else {
result.completeExceptionally(errors.first())
}
}
}
return result.thenAccept {
log.info {
"RemoteBuildCacheServer has been gracefully shut down"
}
}
}
}
fun sendShutdownSignal() {
bossGroup.shutdownGracefully()
}
}
fun run(): ServerHandle {
// Create the multithreaded event loops for the server
val bossGroup = MultiThreadIoEventLoopGroup(1, NioIoHandler.newFactory())
val channelFactory = ChannelFactory<SocketChannel> { NioSocketChannel() }
val datagramChannelFactory = ChannelFactory<DatagramChannel> { NioDatagramChannel() }
val serverChannelFactory = ChannelFactory<ServerSocketChannel> { NioServerSocketChannel() }
val workerGroup = MultiThreadIoEventLoopGroup(0, NioIoHandler.newFactory())
val serverInitializer = ServerInitializer(cfg, channelFactory, datagramChannelFactory)
val bootstrap = ServerBootstrap().apply {
// Configure the server
group(bossGroup, workerGroup)
channelFactory(serverChannelFactory)
childHandler(serverInitializer)
option(ChannelOption.SO_BACKLOG, cfg.incomingConnectionsBacklogSize)
childOption(ChannelOption.SO_KEEPALIVE, true)
}
// Bind and start to accept incoming connections.
val bindAddress = InetSocketAddress(cfg.host, cfg.port)
val httpChannel = bootstrap.bind(bindAddress).sync().channel()
log.info {
"RemoteBuildCacheServer is listening on ${cfg.host}:${cfg.port}"
}
return ServerHandle(
httpChannel.closeFuture(),
bossGroup,
setOf(workerGroup),
serverInitializer
)
}
}
@@ -0,0 +1,88 @@
package net.woggioni.rbcs.server.auth
import io.netty.buffer.Unpooled
import io.netty.channel.ChannelFutureListener
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter
import io.netty.handler.codec.http.DefaultFullHttpResponse
import io.netty.handler.codec.http.FullHttpResponse
import io.netty.handler.codec.http.HttpContent
import io.netty.handler.codec.http.HttpHeaderNames
import io.netty.handler.codec.http.HttpRequest
import io.netty.handler.codec.http.HttpResponseStatus
import io.netty.handler.codec.http.HttpVersion
import io.netty.util.ReferenceCountUtil
import java.net.InetSocketAddress
import net.woggioni.rbcs.api.Configuration
import net.woggioni.rbcs.api.Configuration.Group
import net.woggioni.rbcs.api.Role
import net.woggioni.rbcs.common.createLogger
import net.woggioni.rbcs.server.RemoteBuildCacheServer
abstract class AbstractNettyHttpAuthenticator(private val authorizer: Authorizer) : ChannelInboundHandlerAdapter() {
companion object {
private val log = createLogger<AbstractNettyHttpAuthenticator>()
private val AUTHENTICATION_FAILED: FullHttpResponse = DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, HttpResponseStatus.UNAUTHORIZED, Unpooled.EMPTY_BUFFER
).apply {
headers()[HttpHeaderNames.CONTENT_LENGTH] = "0"
}
private val NOT_AUTHORIZED: FullHttpResponse = DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, HttpResponseStatus.FORBIDDEN, Unpooled.EMPTY_BUFFER
).apply {
headers()[HttpHeaderNames.CONTENT_LENGTH] = "0"
}
}
class AuthenticationResult(val user: Configuration.User?, val groups: Set<Group>)
abstract fun authenticate(ctx: ChannelHandlerContext, req: HttpRequest): AuthenticationResult?
override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
if (msg is HttpRequest) {
val result = authenticate(ctx, msg) ?: return authenticationFailure(ctx, msg)
ctx.channel().attr(RemoteBuildCacheServer.userAttribute).set(result.user)
ctx.channel().attr(RemoteBuildCacheServer.groupAttribute).set(result.groups)
val roles = (
(result.user?.let { user ->
user.groups.asSequence().flatMap { group ->
group.roles.asSequence()
}
} ?: emptySequence<Role>()) +
result.groups.asSequence().flatMap { it.roles.asSequence() }
).toSet()
val authorized = authorizer.authorize(roles, msg)
if(log.isDebugEnabled) {
val authorizedMessage = if(authorized) { "Authorized" } else { "Forbidden" }
val clientAddress = ctx.channel().attr<InetSocketAddress>(RemoteBuildCacheServer.clientIp).get()
val roleString = "[" + roles.asSequence().map { "\"" + it + "\""}.joinToString(", ") + "]"
result.user?.name?.takeUnless(String::isEmpty)?.let { username ->
log.debug("$authorizedMessage ${msg.method()} request from user $username with address $clientAddress, granted roles $roleString")
} ?: {
log.debug("$authorizedMessage anonymous ${msg.method()} request with address $clientAddress, granted roles $roleString")
}
}
if (authorized) {
super.channelRead(ctx, msg)
} else {
authorizationFailure(ctx, msg)
}
} else if(msg is HttpContent) {
ctx.fireChannelRead(msg)
}
}
private fun authenticationFailure(ctx: ChannelHandlerContext, msg: Any) {
ReferenceCountUtil.release(msg)
ctx.writeAndFlush(AUTHENTICATION_FAILED.retainedDuplicate()).addListener(ChannelFutureListener.CLOSE_ON_FAILURE)
}
private fun authorizationFailure(ctx: ChannelHandlerContext, msg: Any) {
ReferenceCountUtil.release(msg)
ctx.writeAndFlush(NOT_AUTHORIZED.retainedDuplicate()).addListener(ChannelFutureListener.CLOSE_ON_FAILURE)
}
}
@@ -0,0 +1,8 @@
package net.woggioni.rbcs.server.auth
import io.netty.handler.codec.http.HttpRequest
import net.woggioni.rbcs.api.Role
fun interface Authorizer {
fun authorize(roles : Set<Role>, request: HttpRequest) : Boolean
}
@@ -0,0 +1,24 @@
package net.woggioni.rbcs.server.auth
import io.netty.handler.codec.http.HttpMethod
import io.netty.handler.codec.http.HttpRequest
import net.woggioni.rbcs.api.Role
class RoleAuthorizer : Authorizer {
companion object {
private val METHOD_MAP = mapOf(
Role.Reader to setOf(HttpMethod.GET, HttpMethod.HEAD),
Role.Writer to setOf(HttpMethod.PUT, HttpMethod.POST),
Role.Healthcheck to setOf(HttpMethod.TRACE)
)
}
override fun authorize(roles: Set<Role>, request: HttpRequest) : Boolean {
val allowedMethods = roles.asSequence()
.mapNotNull(METHOD_MAP::get)
.flatten()
.toSet()
return request.method() in allowedMethods
}
}
@@ -0,0 +1,169 @@
package net.woggioni.rbcs.server.cache
import java.io.ByteArrayOutputStream
import java.io.InputStream
import java.io.ObjectInputStream
import java.io.ObjectOutputStream
import java.io.Serializable
import java.nio.ByteBuffer
import java.nio.channels.Channels
import java.nio.channels.FileChannel
import java.nio.file.Files
import java.nio.file.Path
import java.nio.file.StandardCopyOption
import java.nio.file.StandardOpenOption
import java.nio.file.attribute.BasicFileAttributes
import java.time.Duration
import java.time.Instant
import java.util.concurrent.CompletableFuture
import net.woggioni.jwo.JWO
import net.woggioni.rbcs.api.AsyncCloseable
import net.woggioni.rbcs.api.CacheValueMetadata
import net.woggioni.rbcs.common.createLogger
class FileSystemCache(
val root: Path,
val maxAge: Duration
) : AsyncCloseable {
class EntryValue(val metadata: CacheValueMetadata, val channel : FileChannel, val offset : Long, val size : Long) : Serializable
private companion object {
private val log = createLogger<FileSystemCache>()
}
init {
Files.createDirectories(root)
}
@Volatile
private var running = true
private var nextGc = Instant.now()
fun get(key: String): EntryValue? =
root.resolve(key).takeIf(Files::exists)
?.let { file ->
val size = Files.size(file)
val channel = FileChannel.open(file, StandardOpenOption.READ)
val source = Channels.newInputStream(channel)
val tmp = ByteArray(Integer.BYTES)
val buffer = ByteBuffer.wrap(tmp)
source.read(tmp)
buffer.rewind()
val offset = (Integer.BYTES + buffer.getInt()).toLong()
var count = 0
val wrapper = object : InputStream() {
override fun read(): Int {
return source.read().also {
if (it > 0) count += it
}
}
override fun read(b: ByteArray, off: Int, len: Int): Int {
return source.read(b, off, len).also {
if (it > 0) count += it
}
}
override fun close() {
}
}
val metadata = ObjectInputStream(wrapper).use { ois ->
ois.readObject() as CacheValueMetadata
}
EntryValue(metadata, channel, offset, size)
}
class FileSink(metadata: CacheValueMetadata, private val path: Path, private val tmpFile: Path) {
val channel: FileChannel
init {
val baos = ByteArrayOutputStream()
ObjectOutputStream(baos).use {
it.writeObject(metadata)
}
Files.newOutputStream(tmpFile).use {
val bytes = baos.toByteArray()
val buffer = ByteBuffer.allocate(Integer.BYTES)
buffer.putInt(bytes.size)
buffer.rewind()
it.write(buffer.array())
it.write(bytes)
}
channel = FileChannel.open(tmpFile, StandardOpenOption.APPEND)
}
fun commit() {
channel.close()
Files.move(tmpFile, path, StandardCopyOption.ATOMIC_MOVE)
}
fun rollback() {
channel.close()
Files.delete(path)
}
}
fun put(
key: String,
metadata: CacheValueMetadata,
): FileSink {
val file = root.resolve(key)
val tmpFile = Files.createTempFile(root, null, ".tmp")
return FileSink(metadata, file, tmpFile)
}
private val closeFuture = object : CompletableFuture<Void>() {
init {
Thread.ofVirtual().name("file-system-cache-gc").start {
try {
while (running) {
gc()
}
complete(null)
} catch (ex : Throwable) {
completeExceptionally(ex)
}
}
}
}
private fun gc() {
val now = Instant.now()
if (nextGc < now) {
val oldestEntry = actualGc(now)
nextGc = (oldestEntry ?: now).plus(maxAge)
}
Thread.sleep(minOf(Duration.between(now, nextGc), Duration.ofSeconds(1)))
}
/**
* Returns the creation timestamp of the oldest cache entry (if any)
*/
private fun actualGc(now: Instant): Instant? {
var result: Instant? = null
Files.list(root)
.filter { path ->
JWO.splitExtension(path)
.map { it._2 }
.map { it != ".tmp" }
.orElse(true)
}
.filter {
val creationTimeStamp = Files.readAttributes(it, BasicFileAttributes::class.java)
.creationTime()
.toInstant()
if (result == null || creationTimeStamp < result) {
result = creationTimeStamp
}
now > creationTimeStamp.plus(maxAge)
}.forEach(Files::delete)
return result
}
override fun asyncClose() : CompletableFuture<Void> {
running = false
return closeFuture
}
}
@@ -0,0 +1,38 @@
package net.woggioni.rbcs.server.cache
import io.netty.channel.ChannelFactory
import io.netty.channel.EventLoopGroup
import io.netty.channel.socket.DatagramChannel
import io.netty.channel.socket.SocketChannel
import java.nio.file.Path
import java.time.Duration
import net.woggioni.jwo.Application
import net.woggioni.rbcs.api.CacheHandlerFactory
import net.woggioni.rbcs.api.Configuration
import net.woggioni.rbcs.common.RBCS
data class FileSystemCacheConfiguration(
val root: Path?,
val maxAge: Duration,
val digestAlgorithm : String?,
val compressionEnabled: Boolean,
val compressionLevel: Int,
) : Configuration.Cache {
override fun materialize() = object : CacheHandlerFactory {
private val cache = FileSystemCache(root ?: Application.builder("rbcs").build().computeCacheDirectory(), maxAge)
override fun asyncClose() = cache.asyncClose()
override fun newHandler(
cfg : Configuration,
eventLoop: EventLoopGroup,
socketChannelFactory: ChannelFactory<SocketChannel>,
datagramChannelFactory: ChannelFactory<DatagramChannel>
) = FileSystemCacheHandler(cache, digestAlgorithm, compressionEnabled, compressionLevel, cfg.connection.chunkSize)
}
override fun getNamespaceURI() = RBCS.RBCS_NAMESPACE_URI
override fun getTypeName() = "fileSystemCacheType"
}
@@ -0,0 +1,137 @@
package net.woggioni.rbcs.server.cache
import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandlerContext
import io.netty.handler.codec.http.LastHttpContent
import io.netty.handler.stream.ChunkedNioFile
import java.nio.channels.Channels
import java.util.Base64
import java.util.zip.Deflater
import java.util.zip.DeflaterOutputStream
import java.util.zip.InflaterInputStream
import net.woggioni.rbcs.api.CacheHandler
import net.woggioni.rbcs.api.message.CacheMessage
import net.woggioni.rbcs.api.message.CacheMessage.CacheContent
import net.woggioni.rbcs.api.message.CacheMessage.CacheGetRequest
import net.woggioni.rbcs.api.message.CacheMessage.CachePutRequest
import net.woggioni.rbcs.api.message.CacheMessage.CachePutResponse
import net.woggioni.rbcs.api.message.CacheMessage.CacheValueFoundResponse
import net.woggioni.rbcs.api.message.CacheMessage.CacheValueNotFoundResponse
import net.woggioni.rbcs.api.message.CacheMessage.LastCacheContent
import net.woggioni.rbcs.common.RBCS.processCacheKey
class FileSystemCacheHandler(
private val cache: FileSystemCache,
private val digestAlgorithm: String?,
private val compressionEnabled: Boolean,
private val compressionLevel: Int,
private val chunkSize: Int
) : CacheHandler() {
private interface InProgressRequest{
}
private class InProgressGetRequest(val request : CacheGetRequest) : InProgressRequest
private inner class InProgressPutRequest(
val key : String,
private val fileSink : FileSystemCache.FileSink
) : InProgressRequest {
private val stream = Channels.newOutputStream(fileSink.channel).let {
if (compressionEnabled) {
DeflaterOutputStream(it, Deflater(compressionLevel))
} else {
it
}
}
fun write(buf: ByteBuf) {
buf.readBytes(stream, buf.readableBytes())
}
fun commit() {
stream.close()
fileSink.commit()
}
fun rollback() {
fileSink.rollback()
}
}
private var inProgressRequest: InProgressRequest? = null
override fun channelRead0(ctx: ChannelHandlerContext, msg: CacheMessage) {
when (msg) {
is CacheGetRequest -> handleGetRequest(ctx, msg)
is CachePutRequest -> handlePutRequest(ctx, msg)
is LastCacheContent -> handleLastCacheContent(ctx, msg)
is CacheContent -> handleCacheContent(ctx, msg)
else -> ctx.fireChannelRead(msg)
}
}
private fun handleGetRequest(ctx: ChannelHandlerContext, msg: CacheGetRequest) {
inProgressRequest = InProgressGetRequest(msg)
}
private fun handlePutRequest(ctx: ChannelHandlerContext, msg: CachePutRequest) {
val key = String(Base64.getUrlEncoder().encode(processCacheKey(msg.key, null, digestAlgorithm)))
val sink = cache.put(key, msg.metadata)
inProgressRequest = InProgressPutRequest(msg.key, sink)
}
private fun handleCacheContent(ctx: ChannelHandlerContext, msg: CacheContent) {
val request = inProgressRequest
if(request is InProgressPutRequest) {
request.write(msg.content())
}
}
private fun handleLastCacheContent(ctx: ChannelHandlerContext, msg: LastCacheContent) {
when(val request = inProgressRequest) {
is InProgressPutRequest -> {
inProgressRequest = null
request.write(msg.content())
request.commit()
sendMessageAndFlush(ctx, CachePutResponse(request.key))
}
is InProgressGetRequest -> {
val key = String(Base64.getUrlEncoder().encode(processCacheKey(request.request.key, null, digestAlgorithm)))
cache.get(key)?.also { entryValue ->
sendMessageAndFlush(ctx, CacheValueFoundResponse(request.request.key, entryValue.metadata))
entryValue.channel.let { channel ->
if(compressionEnabled) {
InflaterInputStream(Channels.newInputStream(channel)).use { stream ->
outerLoop@
while (true) {
val buf = ctx.alloc().heapBuffer(chunkSize)
while(buf.readableBytes() < chunkSize) {
val read = buf.writeBytes(stream, chunkSize)
if(read < 0) {
sendMessageAndFlush(ctx, LastCacheContent(buf))
break@outerLoop
}
}
sendMessageAndFlush(ctx, CacheContent(buf))
}
}
} else {
sendMessage(ctx, ChunkedNioFile(channel, entryValue.offset, entryValue.size - entryValue.offset, chunkSize))
sendMessageAndFlush(ctx, LastHttpContent.EMPTY_LAST_CONTENT)
}
}
} ?: sendMessageAndFlush(ctx, CacheValueNotFoundResponse(key))
}
}
}
override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
(inProgressRequest as? InProgressPutRequest)?.rollback()
super.exceptionCaught(ctx, cause)
}
}
@@ -0,0 +1,65 @@
package net.woggioni.rbcs.server.cache
import java.nio.file.Path
import java.time.Duration
import java.util.zip.Deflater
import net.woggioni.rbcs.api.CacheProvider
import net.woggioni.rbcs.common.RBCS
import net.woggioni.rbcs.common.Xml
import net.woggioni.rbcs.common.Xml.Companion.renderAttribute
import org.w3c.dom.Document
import org.w3c.dom.Element
class FileSystemCacheProvider : CacheProvider<FileSystemCacheConfiguration> {
override fun getXmlSchemaLocation() = "classpath:net/woggioni/rbcs/server/schema/rbcs-server.xsd"
override fun getXmlType() = "fileSystemCacheType"
override fun getXmlNamespace() = "urn:net.woggioni.rbcs.server"
override fun deserialize(el: Element): FileSystemCacheConfiguration {
val path = el.renderAttribute("path")
?.let(Path::of)
val maxAge = el.renderAttribute("max-age")
?.let(Duration::parse)
?: Duration.ofDays(1)
val enableCompression = el.renderAttribute("enable-compression")
?.let(String::toBoolean)
?: true
val compressionLevel = el.renderAttribute("compression-level")
?.let(String::toInt)
?: Deflater.DEFAULT_COMPRESSION
val digestAlgorithm = el.renderAttribute("digest")
return FileSystemCacheConfiguration(
path,
maxAge,
digestAlgorithm,
enableCompression,
compressionLevel,
)
}
override fun serialize(doc: Document, cache : FileSystemCacheConfiguration) = cache.run {
val result = doc.createElement("cache")
Xml.of(doc, result) {
val prefix = doc.lookupPrefix(RBCS.RBCS_NAMESPACE_URI)
attr("xs:type", "${prefix}:fileSystemCacheType", RBCS.XML_SCHEMA_NAMESPACE_URI)
root?.let {
attr("path", it.toString())
}
attr("max-age", maxAge.toString())
digestAlgorithm?.let { digestAlgorithm ->
attr("digest", digestAlgorithm)
}
attr("enable-compression", compressionEnabled.toString())
compressionLevel.takeIf {
it != Deflater.DEFAULT_COMPRESSION
}?.let {
attr("compression-level", it.toString())
}
}
result
}
}
@@ -0,0 +1,131 @@
package net.woggioni.rbcs.server.cache
import java.time.Duration
import java.time.Instant
import java.util.PriorityQueue
import java.util.concurrent.CompletableFuture
import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.ReentrantReadWriteLock
import kotlin.concurrent.withLock
import net.woggioni.rbcs.api.AsyncCloseable
import net.woggioni.rbcs.api.CacheValueMetadata
import net.woggioni.rbcs.common.createLogger
private class CacheKey(private val value: ByteArray) {
override fun equals(other: Any?) = if (other is CacheKey) {
value.contentEquals(other.value)
} else false
override fun hashCode() = value.contentHashCode()
}
class CacheEntry(
val metadata: CacheValueMetadata,
val content: ByteArray
)
class InMemoryCache(
private val maxAge: Duration,
private val maxSize: Long
) : AsyncCloseable {
companion object {
private val log = createLogger<InMemoryCache>()
}
private var mapSize : Long = 0
private val map = HashMap<CacheKey, CacheEntry>()
private val lock = ReentrantReadWriteLock()
private val cond = lock.writeLock().newCondition()
private class RemovalQueueElement(val key: CacheKey, val value: CacheEntry, val expiry: Instant) :
Comparable<RemovalQueueElement> {
override fun compareTo(other: RemovalQueueElement) = expiry.compareTo(other.expiry)
}
private val removalQueue = PriorityQueue<RemovalQueueElement>()
@Volatile
private var running = true
private val closeFuture = object : CompletableFuture<Void>() {
init {
Thread.ofVirtual().name("in-memory-cache-gc").start {
try {
lock.writeLock().withLock {
while (running) {
val el = removalQueue.poll()
if(el == null) {
cond.await(1000, TimeUnit.MILLISECONDS)
continue
}
val value = el.value
val now = Instant.now()
if (now > el.expiry) {
val removed = map.remove(el.key, value)
if (removed) {
updateSizeAfterRemoval(value.content)
}
} else {
removalQueue.offer(el)
val interval = minOf(Duration.between(now, el.expiry), Duration.ofSeconds(1))
cond.await(interval.toMillis(), TimeUnit.MILLISECONDS)
}
}
map.clear()
}
complete(null)
} catch (ex: Throwable) {
completeExceptionally(ex)
}
}
}
}
fun removeEldest(): Long {
while (true) {
val el = removalQueue.poll() ?: return mapSize
val value = el.value
val removed = map.remove(el.key, value)
if (removed) {
val newSize = updateSizeAfterRemoval(value.content)
return newSize
}
}
}
private fun updateSizeAfterRemoval(removed: ByteArray): Long {
mapSize -= removed.size
return mapSize
}
override fun asyncClose() : CompletableFuture<Void> {
running = false
lock.writeLock().withLock {
cond.signal()
}
return closeFuture
}
fun get(key: ByteArray) = lock.readLock().withLock {
map[CacheKey(key)]?.run {
CacheEntry(metadata, content)
}
}
fun put(
key: ByteArray,
value: CacheEntry,
) {
val cacheKey = CacheKey(key)
lock.writeLock().withLock {
val oldSize = map.put(cacheKey, value)?.content?.size ?: 0
val delta = value.content.size - oldSize
mapSize += delta
removalQueue.offer(RemovalQueueElement(cacheKey, value, Instant.now().plus(maxAge)))
while (mapSize > maxSize) {
removeEldest()
}
}
}
}
@@ -0,0 +1,35 @@
package net.woggioni.rbcs.server.cache
import io.netty.channel.ChannelFactory
import io.netty.channel.EventLoopGroup
import io.netty.channel.socket.DatagramChannel
import io.netty.channel.socket.SocketChannel
import java.time.Duration
import net.woggioni.rbcs.api.CacheHandlerFactory
import net.woggioni.rbcs.api.Configuration
import net.woggioni.rbcs.common.RBCS
data class InMemoryCacheConfiguration(
val maxAge: Duration,
val maxSize: Long,
val digestAlgorithm : String?,
val compressionEnabled: Boolean,
val compressionLevel: Int,
) : Configuration.Cache {
override fun materialize() = object : CacheHandlerFactory {
private val cache = InMemoryCache(maxAge, maxSize)
override fun asyncClose() = cache.asyncClose()
override fun newHandler(
cfg : Configuration,
eventLoop: EventLoopGroup,
socketChannelFactory: ChannelFactory<SocketChannel>,
datagramChannelFactory: ChannelFactory<DatagramChannel>
) = InMemoryCacheHandler(cache, digestAlgorithm, compressionEnabled, compressionLevel)
}
override fun getNamespaceURI() = RBCS.RBCS_NAMESPACE_URI
override fun getTypeName() = "inMemoryCacheType"
}
@@ -0,0 +1,155 @@
package net.woggioni.rbcs.server.cache
import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandlerContext
import java.util.zip.Deflater
import java.util.zip.DeflaterOutputStream
import java.util.zip.InflaterOutputStream
import net.woggioni.rbcs.api.CacheHandler
import net.woggioni.rbcs.api.message.CacheMessage
import net.woggioni.rbcs.api.message.CacheMessage.CacheContent
import net.woggioni.rbcs.api.message.CacheMessage.CacheGetRequest
import net.woggioni.rbcs.api.message.CacheMessage.CachePutRequest
import net.woggioni.rbcs.api.message.CacheMessage.CachePutResponse
import net.woggioni.rbcs.api.message.CacheMessage.CacheValueFoundResponse
import net.woggioni.rbcs.api.message.CacheMessage.CacheValueNotFoundResponse
import net.woggioni.rbcs.api.message.CacheMessage.LastCacheContent
import net.woggioni.rbcs.common.ByteBufOutputStream
import net.woggioni.rbcs.common.RBCS.processCacheKey
class InMemoryCacheHandler(
private val cache: InMemoryCache,
private val digestAlgorithm: String?,
private val compressionEnabled: Boolean,
private val compressionLevel: Int
) : CacheHandler() {
private interface InProgressRequest : AutoCloseable {
}
private class InProgressGetRequest(val request: CacheGetRequest) : InProgressRequest {
override fun close() {
}
}
private interface InProgressPutRequest : InProgressRequest {
val request: CachePutRequest
val buf: ByteBuf
fun append(buf: ByteBuf)
}
private inner class InProgressPlainPutRequest(ctx: ChannelHandlerContext, override val request: CachePutRequest) :
InProgressPutRequest {
override val buf = ctx.alloc().compositeHeapBuffer()
override fun append(buf: ByteBuf) {
if (buf.isDirect) {
this.buf.writeBytes(buf)
} else {
this.buf.addComponent(true, buf.retain())
}
}
override fun close() {
buf.release()
}
}
private inner class InProgressCompressedPutRequest(
ctx: ChannelHandlerContext,
override val request: CachePutRequest
) : InProgressPutRequest {
override val buf = ctx.alloc().heapBuffer()
private val stream = ByteBufOutputStream(buf).let {
DeflaterOutputStream(it, Deflater(compressionLevel))
}
override fun append(buf: ByteBuf) {
buf.readBytes(stream, buf.readableBytes())
}
override fun close() {
stream.close()
}
}
private var inProgressRequest: InProgressRequest? = null
override fun channelRead0(ctx: ChannelHandlerContext, msg: CacheMessage) {
when (msg) {
is CacheGetRequest -> handleGetRequest(ctx, msg)
is CachePutRequest -> handlePutRequest(ctx, msg)
is LastCacheContent -> handleLastCacheContent(ctx, msg)
is CacheContent -> handleCacheContent(ctx, msg)
else -> ctx.fireChannelRead(msg)
}
}
private fun handleGetRequest(ctx: ChannelHandlerContext, msg: CacheGetRequest) {
inProgressRequest = InProgressGetRequest(msg)
}
private fun handlePutRequest(ctx: ChannelHandlerContext, msg: CachePutRequest) {
inProgressRequest = if (compressionEnabled) {
InProgressCompressedPutRequest(ctx, msg)
} else {
InProgressPlainPutRequest(ctx, msg)
}
}
private fun handleCacheContent(ctx: ChannelHandlerContext, msg: CacheContent) {
val req = inProgressRequest
if (req is InProgressPutRequest) {
req.append(msg.content())
}
}
private fun handleLastCacheContent(ctx: ChannelHandlerContext, msg: LastCacheContent) {
handleCacheContent(ctx, msg)
when (val req = inProgressRequest) {
is InProgressGetRequest -> {
// this.inProgressRequest = null
cache.get(processCacheKey(req.request.key, null, digestAlgorithm))?.let { value ->
sendMessageAndFlush(ctx, CacheValueFoundResponse(req.request.key, value.metadata))
if (compressionEnabled) {
val buf = ctx.alloc().heapBuffer()
InflaterOutputStream(ByteBufOutputStream(buf)).use {
it.write(value.content)
buf.retain()
}
sendMessage(ctx, LastCacheContent(buf))
} else {
val buf = ctx.alloc().heapBuffer()
ByteBufOutputStream(buf).use {
it.write(value.content)
buf.retain()
}
sendMessage(ctx, LastCacheContent(buf))
}
} ?: sendMessage(ctx, CacheValueNotFoundResponse(req.request.key))
}
is InProgressPutRequest -> {
this.inProgressRequest = null
val buf = req.buf
buf.retain()
req.close()
val bytes = ByteArray(buf.readableBytes()).also(buf::readBytes)
buf.release()
val cacheKey = processCacheKey(req.request.key, null, digestAlgorithm)
cache.put(cacheKey, CacheEntry(req.request.metadata, bytes))
sendMessageAndFlush(ctx, CachePutResponse(req.request.key))
}
}
}
override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
inProgressRequest?.close()
inProgressRequest = null
super.exceptionCaught(ctx, cause)
}
}
@@ -0,0 +1,62 @@
package net.woggioni.rbcs.server.cache
import java.time.Duration
import java.util.zip.Deflater
import net.woggioni.rbcs.api.CacheProvider
import net.woggioni.rbcs.common.RBCS
import net.woggioni.rbcs.common.Xml
import net.woggioni.rbcs.common.Xml.Companion.renderAttribute
import org.w3c.dom.Document
import org.w3c.dom.Element
class InMemoryCacheProvider : CacheProvider<InMemoryCacheConfiguration> {
override fun getXmlSchemaLocation() = "classpath:net/woggioni/rbcs/server/schema/rbcs-server.xsd"
override fun getXmlType() = "inMemoryCacheType"
override fun getXmlNamespace() = "urn:net.woggioni.rbcs.server"
override fun deserialize(el: Element): InMemoryCacheConfiguration {
val maxAge = el.renderAttribute("max-age")
?.let(Duration::parse)
?: Duration.ofDays(1)
val maxSize = el.renderAttribute("max-size")
?.let(java.lang.Long::decode)
?: 0x1000000
val enableCompression = el.renderAttribute("enable-compression")
?.let(String::toBoolean)
?: true
val compressionLevel = el.renderAttribute("compression-level")
?.let(String::toInt)
?: Deflater.DEFAULT_COMPRESSION
val digestAlgorithm = el.renderAttribute("digest")
return InMemoryCacheConfiguration(
maxAge,
maxSize,
digestAlgorithm,
enableCompression,
compressionLevel,
)
}
override fun serialize(doc: Document, cache : InMemoryCacheConfiguration) = cache.run {
val result = doc.createElement("cache")
Xml.of(doc, result) {
val prefix = doc.lookupPrefix(RBCS.RBCS_NAMESPACE_URI)
attr("xs:type", "${prefix}:inMemoryCacheType", RBCS.XML_SCHEMA_NAMESPACE_URI)
attr("max-age", maxAge.toString())
attr("max-size", maxSize.toString())
digestAlgorithm?.let { digestAlgorithm ->
attr("digest", digestAlgorithm)
}
attr("enable-compression", compressionEnabled.toString())
compressionLevel.takeIf {
it != Deflater.DEFAULT_COMPRESSION
}?.let {
attr("compression-level", it.toString())
}
}
result
}
}
@@ -0,0 +1,15 @@
package net.woggioni.rbcs.server.configuration
import java.util.ServiceLoader
import net.woggioni.rbcs.api.CacheProvider
import net.woggioni.rbcs.api.Configuration
object CacheSerializers {
val index = (Configuration::class.java.module.layer?.let { layer ->
ServiceLoader.load(layer, CacheProvider::class.java)
} ?: ServiceLoader.load(CacheProvider::class.java))
.asSequence()
.map {
(it.xmlNamespace to it.xmlType) to it
}.toMap()
}
@@ -0,0 +1,361 @@
package net.woggioni.rbcs.server.configuration
import java.nio.file.Paths
import java.time.Duration
import java.time.temporal.ChronoUnit
import net.woggioni.rbcs.api.Configuration
import net.woggioni.rbcs.api.Configuration.Authentication
import net.woggioni.rbcs.api.Configuration.BasicAuthentication
import net.woggioni.rbcs.api.Configuration.Cache
import net.woggioni.rbcs.api.Configuration.ClientCertificateAuthentication
import net.woggioni.rbcs.api.Configuration.ForwardedClientCertificateAuthentication
import net.woggioni.rbcs.api.Configuration.Group
import net.woggioni.rbcs.api.Configuration.KeyStore
import net.woggioni.rbcs.api.Configuration.Tls
import net.woggioni.rbcs.api.Configuration.TlsCertificateExtractor
import net.woggioni.rbcs.api.Configuration.TrustStore
import net.woggioni.rbcs.api.Configuration.User
import net.woggioni.rbcs.api.Role
import net.woggioni.rbcs.api.exception.ConfigurationException
import net.woggioni.rbcs.common.Cidr
import net.woggioni.rbcs.common.Xml.Companion.asIterable
import net.woggioni.rbcs.common.Xml.Companion.renderAttribute
import org.w3c.dom.Document
import org.w3c.dom.Element
import org.w3c.dom.TypeInfo
object Parser {
fun parse(document: Document): Configuration {
val root = document.documentElement
val anonymousUser = User("", null, emptySet(), null)
var connection: Configuration.Connection = Configuration.Connection(
Duration.of(30, ChronoUnit.SECONDS),
Duration.of(60, ChronoUnit.SECONDS),
Duration.of(60, ChronoUnit.SECONDS),
0x4000000,
0x10000
)
var rateLimiter = Configuration.RateLimiter(false, 0x100000, 100)
var eventExecutor: Configuration.EventExecutor = Configuration.EventExecutor(true)
var cache: Cache? = null
var host = "127.0.0.1"
var port = 11080
var proxyProtocolEnabled = false
var trustedProxies = emptyList<Cidr>()
var users: Map<String, User> = mapOf(anonymousUser.name to anonymousUser)
var groups = emptyMap<String, Group>()
var tls: Tls? = null
val serverPath = root.renderAttribute("path")
var incomingConnectionsBacklogSize = 1024
var authentication: Authentication? = null
for (child in root.asIterable()) {
val tagName = child.localName
when (tagName) {
"authentication" -> {
for (gchild in child.asIterable()) {
when (gchild.localName) {
"basic" -> {
authentication = BasicAuthentication()
}
"client-certificate" -> {
var tlsExtractorUser: TlsCertificateExtractor? = null
var tlsExtractorGroup: TlsCertificateExtractor? = null
for (ggchild in gchild.asIterable()) {
when (ggchild.localName) {
"group-extractor" -> {
val attrName = ggchild.renderAttribute("attribute-name")
val pattern = ggchild.renderAttribute("pattern")
tlsExtractorGroup = TlsCertificateExtractor(attrName, pattern)
}
"user-extractor" -> {
val attrName = ggchild.renderAttribute("attribute-name")
val pattern = ggchild.renderAttribute("pattern")
tlsExtractorUser = TlsCertificateExtractor(attrName, pattern)
}
}
}
authentication = ClientCertificateAuthentication(tlsExtractorUser, tlsExtractorGroup)
}
"forwarded-client-certificate" -> {
val headerName = gchild.renderAttribute("header-name") ?: "X-Client-Cert-Subject-DN"
var tlsExtractorUser: TlsCertificateExtractor? = null
var tlsExtractorGroup: TlsCertificateExtractor? = null
for (ggchild in gchild.asIterable()) {
when (ggchild.localName) {
"group-extractor" -> {
val attrName = ggchild.renderAttribute("attribute-name")
val pattern = ggchild.renderAttribute("pattern")
tlsExtractorGroup = TlsCertificateExtractor(attrName, pattern)
}
"user-extractor" -> {
val attrName = ggchild.renderAttribute("attribute-name")
val pattern = ggchild.renderAttribute("pattern")
tlsExtractorUser = TlsCertificateExtractor(attrName, pattern)
}
}
}
authentication = ForwardedClientCertificateAuthentication(headerName, tlsExtractorUser, tlsExtractorGroup)
}
}
}
}
"authorization" -> {
var knownUsers = sequenceOf(anonymousUser)
for (gchild in child.asIterable()) {
when (gchild.localName) {
"users" -> {
knownUsers += parseUsers(gchild)
}
"groups" -> {
val pair = parseGroups(gchild, knownUsers)
users = pair.first
groups = pair.second
}
}
}
}
"bind" -> {
host = child.renderAttribute("host") ?: throw ConfigurationException("host attribute is required")
port = Integer.parseInt(child.renderAttribute("port"))
proxyProtocolEnabled = child.renderAttribute("proxy-protocol")
?.let(String::toBoolean) ?: false
incomingConnectionsBacklogSize = child.renderAttribute("incoming-connections-backlog-size")
?.let(Integer::parseInt)
?: 1024
for(grandChild in child.asIterable()) {
when(grandChild.localName) {
"trusted-proxies" -> {
trustedProxies = parseTrustedProxies(grandChild)
}
}
}
child.asIterable().filter {
it.localName == "trusted-proxies"
}.firstOrNull()?.let(::parseTrustedProxies)
}
"cache" -> {
cache = (child as TypeInfo).let { tf ->
val typeNamespace = tf.typeNamespace
val typeName = tf.typeName
CacheSerializers.index[typeNamespace to typeName]
?: throw IllegalArgumentException("Cache provider for namespace '$typeNamespace' with name '$typeName' not found")
}.deserialize(child)
}
"connection" -> {
val idleTimeout = child.renderAttribute("idle-timeout")
?.let(Duration::parse) ?: Duration.of(30, ChronoUnit.SECONDS)
val readIdleTimeout = child.renderAttribute("read-idle-timeout")
?.let(Duration::parse) ?: Duration.of(60, ChronoUnit.SECONDS)
val writeIdleTimeout = child.renderAttribute("write-idle-timeout")
?.let(Duration::parse) ?: Duration.of(60, ChronoUnit.SECONDS)
val maxRequestSize = child.renderAttribute("max-request-size")
?.let(Integer::decode) ?: 0x4000000
val chunkSize = child.renderAttribute("chunk-size")
?.let(Integer::decode) ?: 0x10000
connection = Configuration.Connection(
idleTimeout,
readIdleTimeout,
writeIdleTimeout,
maxRequestSize,
chunkSize
)
}
"event-executor" -> {
val useVirtualThread = child.renderAttribute("use-virtual-threads")
?.let(String::toBoolean) ?: true
eventExecutor = Configuration.EventExecutor(useVirtualThread)
}
"rate-limiter" -> {
val delayResponse = child.renderAttribute("delay-response")
?.let(String::toBoolean)
?: false
val messageBufferSize = child.renderAttribute("message-buffer-size")
?.let(Integer::decode)
?: 0x100000
val maxQueuedMessages = child.renderAttribute("max-queued-messages")
?.let(Integer::decode)
?: 100
rateLimiter = Configuration.RateLimiter(delayResponse, messageBufferSize, maxQueuedMessages)
}
"tls" -> {
var keyStore: KeyStore? = null
var trustStore: TrustStore? = null
for (granChild in child.asIterable()) {
when (granChild.localName) {
"keystore" -> {
val keyStoreFile = Paths.get(granChild.renderAttribute("file"))
val keyStorePassword = granChild.renderAttribute("password")
val keyAlias = granChild.renderAttribute("key-alias")
val keyPassword = granChild.renderAttribute("key-password")
keyStore = KeyStore(
keyStoreFile,
keyStorePassword,
keyAlias,
keyPassword
)
}
"truststore" -> {
val trustStoreFile = Paths.get(granChild.renderAttribute("file"))
val trustStorePassword = granChild.renderAttribute("password")
val checkCertificateStatus = granChild.renderAttribute("check-certificate-status")
?.let(String::toBoolean)
?: false
val requireClientCertificate = child.renderAttribute("require-client-certificate")
?.let(String::toBoolean) ?: false
trustStore = TrustStore(
trustStoreFile,
trustStorePassword,
checkCertificateStatus,
requireClientCertificate
)
}
}
}
tls = Tls(keyStore, trustStore)
}
}
}
return Configuration.of(
host,
port,
proxyProtocolEnabled,
trustedProxies,
incomingConnectionsBacklogSize,
serverPath,
eventExecutor,
rateLimiter,
connection,
users,
groups,
cache!!,
authentication,
tls,
)
}
private fun parseRoles(root: Element) = root.asIterable().asSequence().map {
when (it.localName) {
"reader" -> Role.Reader
"writer" -> Role.Writer
"healthcheck" -> Role.Healthcheck
else -> throw UnsupportedOperationException("Illegal node '${it.localName}'")
}
}.toSet()
private fun parseTrustedProxies(root: Element) = root.asIterable().asSequence().map {
when (it.localName) {
"allow" -> it.renderAttribute("cidr")
?.let(Cidr::from)
?: throw ConfigurationException("Missing 'cidr' attribute")
else -> throw ConfigurationException("Unrecognized tag '${it.localName}'")
}
}.toList()
private fun parseUserRefs(root: Element) = root.asIterable().asSequence().map {
when (it.localName) {
"user" -> it.renderAttribute("ref")
"anonymous" -> ""
else -> ConfigurationException("Unrecognized tag '${it.localName}'")
}
}
private fun parseQuota(el: Element): Configuration.Quota {
val calls = el.renderAttribute("calls")
?.let(String::toLong)
?: throw ConfigurationException("Missing attribute 'calls'")
val maxAvailableCalls = el.renderAttribute("max-available-calls")
?.let(String::toLong)
?: calls
val initialAvailableCalls = el.renderAttribute("initial-available-calls")
?.let(String::toLong)
?: maxAvailableCalls
val period = el.renderAttribute("period")
?.let(Duration::parse)
?: throw ConfigurationException("Missing attribute 'period'")
return Configuration.Quota(calls, period, initialAvailableCalls, maxAvailableCalls)
}
private fun parseUsers(root: Element): Sequence<User> {
return root.asIterable().asSequence().mapNotNull { child ->
when (child.localName) {
"user" -> {
val username = child.renderAttribute("name")
val password = child.renderAttribute("password")
var quota: Configuration.Quota? = null
for (gchild in child.asIterable()) {
if (gchild.localName == "quota") {
quota = parseQuota(gchild)
}
}
User(username, password, emptySet(), quota)
}
"anonymous" -> {
var quota: Configuration.Quota? = null
for (gchild in child.asIterable()) {
if (gchild.localName == "quota") {
quota= parseQuota(gchild)
}
}
User("", null, emptySet(), quota)
}
else -> null
}
}
}
private fun parseGroups(root: Element, knownUsers: Sequence<User>): Pair<Map<String, User>, Map<String, Group>> {
val knownUsersMap = knownUsers.associateBy(User::getName)
val userGroups = mutableMapOf<String, MutableSet<String>>()
val groups = root.asIterable().asSequence().filter {
it.localName == "group"
}.map { el ->
val groupName = el.renderAttribute("name") ?: throw ConfigurationException("Group name is required")
var roles = emptySet<Role>()
var userQuota: Configuration.Quota? = null
var groupQuota: Configuration.Quota? = null
for (child in el.asIterable()) {
when (child.localName) {
"users" -> {
parseUserRefs(child).mapNotNull(knownUsersMap::get).forEach { user ->
userGroups.computeIfAbsent(user.name) {
mutableSetOf()
}.add(groupName)
}
}
"roles" -> {
roles = parseRoles(child)
}
"group-quota" -> {
userQuota = parseQuota(child)
}
"user-quota" -> {
groupQuota = parseQuota(child)
}
}
}
groupName to Group(groupName, roles, userQuota, groupQuota)
}.toMap()
val users = knownUsersMap.map { (name, user) ->
name to User(name, user.password, userGroups[name]?.mapNotNull { groups[it] }?.toSet() ?: emptySet(), user.quota)
}.toMap()
return users to groups
}
}
@@ -0,0 +1,218 @@
package net.woggioni.rbcs.server.configuration
import net.woggioni.rbcs.api.CacheProvider
import net.woggioni.rbcs.api.Configuration
import net.woggioni.rbcs.common.RBCS
import net.woggioni.rbcs.common.Xml
import org.w3c.dom.Document
object Serializer {
private fun Xml.serializeQuota(quota : Configuration.Quota) {
attr("calls", quota.calls.toString())
attr("period", quota.period.toString())
attr("max-available-calls", quota.maxAvailableCalls.toString())
attr("initial-available-calls", quota.initialAvailableCalls.toString())
}
fun serialize(conf : Configuration) : Document {
val schemaLocations = CacheSerializers.index.values.asSequence().map {
it.xmlNamespace to it.xmlSchemaLocation
}.toMap()
return Xml.of(RBCS.RBCS_NAMESPACE_URI, RBCS.RBCS_PREFIX + ":server") {
// attr("xmlns:xs", GradleBuildCacheServer.XML_SCHEMA_NAMESPACE_URI)
val value = schemaLocations.asSequence().map { (k, v) -> "$k $v" }.joinToString(" ")
attr("xs:schemaLocation", value , namespaceURI = RBCS.XML_SCHEMA_NAMESPACE_URI)
conf.serverPath
?.takeIf(String::isNotEmpty)
?.let { serverPath ->
attr("path", serverPath)
}
node("bind") {
attr("host", conf.host)
attr("port", conf.port.toString())
attr("incoming-connections-backlog-size", conf.incomingConnectionsBacklogSize.toString())
attr("proxy-protocol", conf.isProxyProtocolEnabled.toString())
if (conf.trustedProxyIPs.isNotEmpty()) {
node("trusted-proxies") {
for(trustedProxy in conf.trustedProxyIPs) {
node("allow") {
attr("cidr", trustedProxy.toString())
}
}
}
}
}
node("connection") {
conf.connection.let { connection ->
attr("idle-timeout", connection.idleTimeout.toString())
attr("read-idle-timeout", connection.readIdleTimeout.toString())
attr("write-idle-timeout", connection.writeIdleTimeout.toString())
attr("max-request-size", connection.maxRequestSize.toString())
attr("chunk-size", connection.chunkSize.toString())
}
}
node("event-executor") {
attr("use-virtual-threads", conf.eventExecutor.isUseVirtualThreads.toString())
}
node("rate-limiter") {
attr("delay-response", conf.rateLimiter.isDelayRequest.toString())
attr("max-queued-messages", conf.rateLimiter.maxQueuedMessages.toString())
attr("message-buffer-size", conf.rateLimiter.messageBufferSize.toString())
}
val cache = conf.cache
val serializer : CacheProvider<Configuration.Cache> =
(CacheSerializers.index[cache.namespaceURI to cache.typeName] as? CacheProvider<Configuration.Cache>) ?: throw NotImplementedError()
element.appendChild(serializer.serialize(doc, cache))
node("authorization") {
node("users") {
for(user in conf.users.values) {
if(user.name.isNotEmpty()) {
node("user") {
attr("name", user.name)
user.password?.let { password ->
attr("password", password)
}
user.quota?.let { quota ->
node("quota") {
serializeQuota(quota)
}
}
}
}
}
conf.users[""]
?.let { anonymousUser ->
anonymousUser.quota?.let { quota ->
node("anonymous") {
node("quota") {
serializeQuota(quota)
}
}
}
}
}
node("groups") {
val groups = conf.users.values.asSequence()
.flatMap {
user -> user.groups.map { it to user }
}.groupBy(Pair<Configuration.Group, Configuration.User>::first, Pair<Configuration.Group, Configuration.User>::second)
for(pair in groups) {
val group = pair.key
val users = pair.value
node("group") {
attr("name", group.name)
if(users.isNotEmpty()) {
node("users") {
var anonymousUser : Configuration.User? = null
for(user in users) {
if(user.name.isNotEmpty()) {
node("user") {
attr("ref", user.name)
}
} else {
anonymousUser = user
}
}
if(anonymousUser != null) {
node("anonymous")
}
}
}
if(group.roles.isNotEmpty()) {
node("roles") {
for(role in group.roles) {
node(role.toString().lowercase())
}
}
}
group.userQuota?.let { quota ->
node("user-quota") {
serializeQuota(quota)
}
}
group.groupQuota?.let { quota ->
node("group-quota") {
serializeQuota(quota)
}
}
}
}
}
}
conf.authentication?.let { authentication ->
node("authentication") {
when(authentication) {
is Configuration.BasicAuthentication -> {
node("basic")
}
is Configuration.ClientCertificateAuthentication -> {
node("client-certificate") {
authentication.groupExtractor?.let { extractor ->
node("group-extractor") {
attr("attribute-name", extractor.rdnType)
attr("pattern", extractor.pattern)
}
}
authentication.userExtractor?.let { extractor ->
node("user-extractor") {
attr("attribute-name", extractor.rdnType)
attr("pattern", extractor.pattern)
}
}
}
}
is Configuration.ForwardedClientCertificateAuthentication -> {
node("forwarded-client-certificate") {
attr("header-name", authentication.headerName)
authentication.groupExtractor?.let { extractor ->
node("group-extractor") {
attr("attribute-name", extractor.rdnType)
attr("pattern", extractor.pattern)
}
}
authentication.userExtractor?.let { extractor ->
node("user-extractor") {
attr("attribute-name", extractor.rdnType)
attr("pattern", extractor.pattern)
}
}
}
}
}
}
}
conf.tls?.let { tlsConfiguration ->
node("tls") {
tlsConfiguration.keyStore?.let { keyStore ->
node("keystore") {
attr("file", keyStore.file.toString())
keyStore.password?.let { keyStorePassword ->
attr("password", keyStorePassword)
}
attr("key-alias", keyStore.keyAlias)
keyStore.keyPassword?.let { keyPassword ->
attr("key-password", keyPassword)
}
}
}
tlsConfiguration.trustStore?.let { trustStore ->
node("truststore") {
attr("file", trustStore.file.toString())
trustStore.password?.let { password ->
attr("password", password)
}
attr("check-certificate-status", trustStore.isCheckCertificateStatus.toString())
attr("require-client-certificate", trustStore.isRequireClientCertificate.toString())
}
}
}
}
}
}
}
@@ -0,0 +1,139 @@
package net.woggioni.rbcs.server.exception
import io.netty.buffer.Unpooled
import io.netty.channel.ChannelDuplexHandler
import io.netty.channel.ChannelFutureListener
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.ChannelHandlerContext
import io.netty.handler.codec.DecoderException
import io.netty.handler.codec.http.DefaultFullHttpResponse
import io.netty.handler.codec.http.FullHttpResponse
import io.netty.handler.codec.http.HttpHeaderNames
import io.netty.handler.codec.http.HttpResponseStatus
import io.netty.handler.codec.http.HttpVersion
import io.netty.handler.timeout.ReadTimeoutException
import io.netty.handler.timeout.WriteTimeoutException
import java.net.ConnectException
import java.net.SocketException
import javax.net.ssl.SSLException
import javax.net.ssl.SSLPeerUnverifiedException
import net.woggioni.rbcs.api.exception.CacheException
import net.woggioni.rbcs.api.exception.ContentTooLargeException
import net.woggioni.rbcs.common.contextLogger
import net.woggioni.rbcs.common.debug
import net.woggioni.rbcs.common.log
import net.woggioni.rbcs.server.RemoteBuildCacheServer
import org.slf4j.event.Level
import org.slf4j.spi.LoggingEventBuilder
@Sharable
object ExceptionHandler : ChannelDuplexHandler() {
val NAME : String = this::class.java.name
private val log = contextLogger()
private val NOT_AUTHORIZED: FullHttpResponse = DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, HttpResponseStatus.FORBIDDEN, Unpooled.EMPTY_BUFFER
).apply {
headers()[HttpHeaderNames.CONTENT_LENGTH] = "0"
}
private val NOT_AVAILABLE: FullHttpResponse = DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, HttpResponseStatus.SERVICE_UNAVAILABLE, Unpooled.EMPTY_BUFFER
).apply {
headers()[HttpHeaderNames.CONTENT_LENGTH] = "0"
}
private val SERVER_ERROR: FullHttpResponse = DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, HttpResponseStatus.INTERNAL_SERVER_ERROR, Unpooled.EMPTY_BUFFER
).apply {
headers()[HttpHeaderNames.CONTENT_LENGTH] = "0"
}
private val TOO_BIG: FullHttpResponse = DefaultFullHttpResponse(
HttpVersion.HTTP_1_1, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, Unpooled.EMPTY_BUFFER
).apply {
headers()[HttpHeaderNames.CONTENT_LENGTH] = "0"
}
override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
when (cause) {
is DecoderException -> {
if(log.isDebugEnabled) {
log.debug(cause.message, cause)
}
ctx.close()
}
is ConnectException -> {
if(log.isErrorEnabled) {
log.error(cause.message, cause)
}
ctx.writeAndFlush(SERVER_ERROR.retainedDuplicate())
}
is SocketException -> {
if(log.isDebugEnabled) {
log.debug(cause.message, cause)
}
ctx.close()
}
is SSLPeerUnverifiedException -> {
if(log.isDebugEnabled) {
log.debug(cause.message, cause)
}
ctx.writeAndFlush(NOT_AUTHORIZED.retainedDuplicate())
.addListener(ChannelFutureListener.CLOSE_ON_FAILURE)
}
is SSLException -> {
if(log.isDebugEnabled) {
log.debug(cause.message, cause)
}
ctx.close()
}
is ContentTooLargeException -> {
log.log(Level.DEBUG, ctx.channel()) { builder : LoggingEventBuilder ->
builder.setMessage("Request body is too large")
}
ctx.writeAndFlush(TOO_BIG.retainedDuplicate())
.addListener(ChannelFutureListener.CLOSE_ON_FAILURE)
}
is ReadTimeoutException -> {
log.debug {
val channelId = ctx.channel().id().asShortText()
"Read timeout on channel $channelId, closing the connection"
}
ctx.close()
}
is WriteTimeoutException -> {
log.debug {
val channelId = ctx.channel().id().asShortText()
"Write timeout on channel $channelId, closing the connection"
}
ctx.close()
}
is CacheException -> {
if(log.isErrorEnabled) {
log.error(cause.message, cause)
}
ctx.writeAndFlush(NOT_AVAILABLE.retainedDuplicate())
.addListener(ChannelFutureListener.CLOSE_ON_FAILURE)
}
else -> {
if(log.isErrorEnabled) {
log.error(cause.message, cause)
}
ctx.writeAndFlush(SERVER_ERROR.retainedDuplicate())
.addListener(ChannelFutureListener.CLOSE_ON_FAILURE)
}
}
}
}
@@ -0,0 +1,13 @@
package net.woggioni.rbcs.server.handler
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.SimpleChannelInboundHandler
import io.netty.handler.codec.http.HttpContent
class BlackHoleRequestHandler : SimpleChannelInboundHandler<HttpContent>() {
companion object {
val NAME = BlackHoleRequestHandler::class.java.name
}
override fun channelRead0(ctx: ChannelHandlerContext, msg: HttpContent) {
}
}
@@ -0,0 +1,40 @@
package net.woggioni.rbcs.server.handler
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter
import io.netty.handler.codec.http.HttpContent
import io.netty.handler.codec.http.HttpRequest
import net.woggioni.rbcs.api.exception.ContentTooLargeException
class MaxRequestSizeHandler(private val maxRequestSize : Int) : ChannelInboundHandlerAdapter() {
companion object {
val NAME = MaxRequestSizeHandler::class.java.name
}
private var cumulativeSize = 0
override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
when(msg) {
is HttpRequest -> {
cumulativeSize = 0
ctx.fireChannelRead(msg)
}
is HttpContent -> {
val exceeded = cumulativeSize > maxRequestSize
if(!exceeded) {
cumulativeSize += msg.content().readableBytes()
}
if(cumulativeSize > maxRequestSize) {
msg.release()
if(!exceeded) {
ctx.fireExceptionCaught(ContentTooLargeException("Request body is too large", null))
}
} else {
ctx.fireChannelRead(msg)
}
}
else -> ctx.fireChannelRead(msg)
}
}
}
@@ -0,0 +1,47 @@
package net.woggioni.rbcs.server.handler
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.SimpleChannelInboundHandler
import io.netty.handler.codec.haproxy.HAProxyMessage
import java.net.InetAddress
import java.net.InetSocketAddress
import net.woggioni.rbcs.common.Cidr
import net.woggioni.rbcs.common.createLogger
import net.woggioni.rbcs.common.trace
import net.woggioni.rbcs.server.RemoteBuildCacheServer
class ProxyProtocolHandler(private val trustedProxyIPs : List<Cidr>) : SimpleChannelInboundHandler<HAProxyMessage>() {
companion object {
private val log = createLogger<ProxyProtocolHandler>()
}
override fun channelRead0(
ctx: ChannelHandlerContext,
msg: HAProxyMessage
) {
val sourceAddress = ctx.channel().remoteAddress()
if (sourceAddress is InetSocketAddress &&
trustedProxyIPs.isEmpty() ||
trustedProxyIPs.any { it.contains((sourceAddress as InetSocketAddress).address) }.also {
if(!it && log.isTraceEnabled) {
log.trace {
"Received a proxied connection request from $sourceAddress which is not a trusted proxy address, " +
"the proxy server address will be used instead"
}
}
}) {
val proxiedClientAddress = InetSocketAddress(
InetAddress.ofLiteral(msg.sourceAddress()),
msg.sourcePort()
)
if(log.isTraceEnabled) {
log.trace {
"Received proxied connection request from $sourceAddress forwarded for $proxiedClientAddress"
}
}
ctx.channel().attr(RemoteBuildCacheServer.clientIp).set(proxiedClientAddress)
}
}
}
@@ -0,0 +1,65 @@
package net.woggioni.rbcs.server.handler
import io.netty.buffer.ByteBufHolder
import io.netty.channel.ChannelDuplexHandler
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelPromise
import io.netty.handler.codec.http.LastHttpContent
import net.woggioni.rbcs.common.createLogger
class ReadTriggerDuplexHandler : ChannelDuplexHandler() {
companion object {
val NAME = ReadTriggerDuplexHandler::class.java.name
private val log = createLogger<ReadTriggerDuplexHandler>()
}
private var inFlight = 0
private val messageBuffer = ArrayDeque<Any>()
override fun handlerAdded(ctx: ChannelHandlerContext) {
ctx.read()
}
override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
if(inFlight > 0) {
messageBuffer.addLast(msg)
} else {
super.channelRead(ctx, msg)
if(msg !is LastHttpContent) {
invokeRead(ctx)
} else {
inFlight += 1
}
}
}
private fun invokeRead(ctx : ChannelHandlerContext) {
if(messageBuffer.isEmpty()) {
ctx.read()
} else {
this.channelRead(ctx, messageBuffer.removeFirst())
}
}
override fun write(
ctx: ChannelHandlerContext,
msg: Any,
promise: ChannelPromise
) {
super.write(ctx, msg, promise)
if(msg is LastHttpContent) {
inFlight -= 1
invokeRead(ctx)
}
}
override fun channelInactive(ctx: ChannelHandlerContext) {
while(messageBuffer.isNotEmpty()) {
val msg = messageBuffer.removeFirst()
if(msg is ByteBufHolder) {
msg.release()
}
}
super.channelInactive(ctx)
}
}
@@ -0,0 +1,227 @@
package net.woggioni.rbcs.server.handler
import io.netty.channel.ChannelDuplexHandler
import io.netty.channel.ChannelHandler
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelPromise
import io.netty.handler.codec.http.DefaultFullHttpResponse
import io.netty.handler.codec.http.DefaultHttpContent
import io.netty.handler.codec.http.DefaultHttpResponse
import io.netty.handler.codec.http.DefaultLastHttpContent
import io.netty.handler.codec.http.HttpContent
import io.netty.handler.codec.http.HttpHeaderNames
import io.netty.handler.codec.http.HttpHeaderValues
import io.netty.handler.codec.http.HttpHeaders
import io.netty.handler.codec.http.HttpMethod
import io.netty.handler.codec.http.HttpRequest
import io.netty.handler.codec.http.HttpResponseStatus
import io.netty.handler.codec.http.HttpUtil
import io.netty.handler.codec.http.HttpVersion
import io.netty.handler.codec.http.LastHttpContent
import java.nio.file.Path
import net.woggioni.rbcs.api.CacheValueMetadata
import net.woggioni.rbcs.api.message.CacheMessage
import net.woggioni.rbcs.api.message.CacheMessage.CacheContent
import net.woggioni.rbcs.api.message.CacheMessage.CacheGetRequest
import net.woggioni.rbcs.api.message.CacheMessage.CachePutRequest
import net.woggioni.rbcs.api.message.CacheMessage.CachePutResponse
import net.woggioni.rbcs.api.message.CacheMessage.CacheValueFoundResponse
import net.woggioni.rbcs.api.message.CacheMessage.CacheValueNotFoundResponse
import net.woggioni.rbcs.api.message.CacheMessage.LastCacheContent
import net.woggioni.rbcs.common.createLogger
import net.woggioni.rbcs.common.debug
import net.woggioni.rbcs.common.warn
import net.woggioni.rbcs.server.exception.ExceptionHandler
class ServerHandler(private val serverPrefix: Path, private val cacheHandlerSupplier : () -> ChannelHandler) :
ChannelDuplexHandler() {
companion object {
private val log = createLogger<ServerHandler>()
val NAME = ServerHandler::class.java.name
}
private var httpVersion = HttpVersion.HTTP_1_1
private var keepAlive = true
private fun resetRequestMetadata() {
httpVersion = HttpVersion.HTTP_1_1
keepAlive = true
}
private fun setRequestMetadata(req: HttpRequest) {
httpVersion = req.protocolVersion()
keepAlive = HttpUtil.isKeepAlive(req)
}
private fun setKeepAliveHeader(headers: HttpHeaders) {
if (!keepAlive) {
headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE)
} else {
headers.set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE)
}
}
private var cacheRequestInProgress : Boolean = false
override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
when (msg) {
is HttpRequest -> handleRequest(ctx, msg)
is HttpContent -> {
if(cacheRequestInProgress) {
if(msg is LastHttpContent) {
super.channelRead(ctx, LastCacheContent(msg.content().retain()))
cacheRequestInProgress = false
} else {
super.channelRead(ctx, CacheContent(msg.content().retain()))
}
msg.release()
} else {
super.channelRead(ctx, msg)
}
}
else -> super.channelRead(ctx, msg)
}
}
override fun write(ctx: ChannelHandlerContext, msg: Any, promise: ChannelPromise?) {
if (msg is CacheMessage) {
try {
when (msg) {
is CachePutResponse -> {
log.debug(ctx) {
"Added value for key '${msg.key}' to build cache"
}
val response = DefaultFullHttpResponse(httpVersion, HttpResponseStatus.CREATED)
val keyBytes = msg.key.toByteArray(Charsets.UTF_8)
response.headers().apply {
set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN)
set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED)
}
setKeepAliveHeader(response.headers())
ctx.write(response)
val buf = ctx.alloc().buffer(keyBytes.size).apply {
writeBytes(keyBytes)
}
ctx.writeAndFlush(DefaultLastHttpContent(buf))
}
is CacheValueNotFoundResponse -> {
log.debug(ctx) {
"Value not found for key '${msg.key}'"
}
val response = DefaultFullHttpResponse(httpVersion, HttpResponseStatus.NOT_FOUND)
response.headers()[HttpHeaderNames.CONTENT_LENGTH] = 0
setKeepAliveHeader(response.headers())
ctx.writeAndFlush(response)
}
is CacheValueFoundResponse -> {
log.debug(ctx) {
"Retrieved value for key '${msg.key}'"
}
val response = DefaultHttpResponse(httpVersion, HttpResponseStatus.OK)
response.headers().apply {
set(HttpHeaderNames.CONTENT_TYPE, msg.metadata.mimeType ?: HttpHeaderValues.APPLICATION_OCTET_STREAM)
msg.metadata.contentDisposition?.let { contentDisposition ->
set(HttpHeaderNames.CONTENT_DISPOSITION, contentDisposition)
}
}
setKeepAliveHeader(response.headers())
response.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED)
ctx.writeAndFlush(response)
}
is LastCacheContent -> {
ctx.writeAndFlush(DefaultLastHttpContent(msg.content()))
}
is CacheContent -> {
ctx.writeAndFlush(DefaultHttpContent(msg.content()))
}
else -> throw UnsupportedOperationException("This should never happen")
}.let { channelFuture ->
if (promise != null) {
channelFuture.addListener {
if (it.isSuccess) promise.setSuccess()
else promise.setFailure(it.cause())
}
}
}
} finally {
resetRequestMetadata()
}
} else if(msg is LastHttpContent) {
ctx.write(msg, promise)
} else super.write(ctx, msg, promise)
}
private fun handleRequest(ctx: ChannelHandlerContext, msg: HttpRequest) {
setRequestMetadata(msg)
val method = msg.method()
if (method === HttpMethod.GET) {
val path = Path.of(msg.uri()).normalize()
if (path.startsWith(serverPrefix)) {
cacheRequestInProgress = true
val relativePath = serverPrefix.relativize(path)
val key : String = relativePath.toString()
val cacheHandler = cacheHandlerSupplier()
ctx.pipeline().addBefore(ExceptionHandler.NAME, null, cacheHandler)
key.let(::CacheGetRequest)
.let(ctx::fireChannelRead)
?: ctx.channel().write(CacheValueNotFoundResponse(key))
} else {
cacheRequestInProgress = false
log.warn(ctx) {
"Got request for unhandled path '${msg.uri()}'"
}
val response = DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.BAD_REQUEST)
response.headers()[HttpHeaderNames.CONTENT_LENGTH] = 0
ctx.writeAndFlush(response)
}
} else if (method === HttpMethod.PUT) {
val path = Path.of(msg.uri()).normalize()
if (path.startsWith(serverPrefix)) {
cacheRequestInProgress = true
val relativePath = serverPrefix.relativize(path)
val key = relativePath.toString()
val cacheHandler = cacheHandlerSupplier()
ctx.pipeline().addAfter(NAME, null, cacheHandler)
path.fileName?.toString()
?.let {
val mimeType = HttpUtil.getMimeType(msg)?.toString()
CachePutRequest(key, CacheValueMetadata(msg.headers().get(HttpHeaderNames.CONTENT_DISPOSITION), mimeType))
}
?.let(ctx::fireChannelRead)
?: ctx.channel().write(CacheValueNotFoundResponse(key))
} else {
cacheRequestInProgress = false
log.warn(ctx) {
"Got request for unhandled path '${msg.uri()}'"
}
val response = DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.BAD_REQUEST)
response.headers()[HttpHeaderNames.CONTENT_LENGTH] = "0"
ctx.writeAndFlush(response)
}
} else if (method == HttpMethod.TRACE) {
cacheRequestInProgress = false
ctx.pipeline().addAfter(NAME, null, TraceHandler)
super.channelRead(ctx, msg)
} else {
cacheRequestInProgress = false
log.warn(ctx) {
"Got request with unhandled method '${msg.method().name()}'"
}
val response = DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.METHOD_NOT_ALLOWED)
response.headers()[HttpHeaderNames.CONTENT_LENGTH] = "0"
ctx.writeAndFlush(response)
}
}
override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
super.exceptionCaught(ctx, cause)
}
}
@@ -0,0 +1,55 @@
package net.woggioni.rbcs.server.handler
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter
import io.netty.handler.codec.http.DefaultHttpResponse
import io.netty.handler.codec.http.HttpContent
import io.netty.handler.codec.http.HttpHeaderNames
import io.netty.handler.codec.http.HttpHeaderValues
import io.netty.handler.codec.http.HttpRequest
import io.netty.handler.codec.http.HttpResponseStatus
import io.netty.handler.codec.http.LastHttpContent
import java.nio.file.Path
@Sharable
object TraceHandler : ChannelInboundHandlerAdapter() {
val NAME = this::class.java.name
override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
when(msg) {
is HttpRequest -> {
val response = DefaultHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK)
response.headers().apply {
set(HttpHeaderNames.CONTENT_TYPE, "message/http")
set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED)
}
ctx.write(response)
val replayedRequestHead = ctx.alloc().buffer()
replayedRequestHead.writeCharSequence(
"TRACE ${Path.of(msg.uri())} ${msg.protocolVersion().text()}\r\n",
Charsets.US_ASCII
)
msg.headers().forEach { (key, value) ->
replayedRequestHead.apply {
writeCharSequence(key, Charsets.US_ASCII)
writeCharSequence(": ", Charsets.US_ASCII)
writeCharSequence(value, Charsets.UTF_8)
writeCharSequence("\r\n", Charsets.US_ASCII)
}
}
replayedRequestHead.writeCharSequence("\r\n", Charsets.US_ASCII)
ctx.writeAndFlush(replayedRequestHead)
}
is LastHttpContent -> {
ctx.writeAndFlush(msg)
ctx.pipeline().remove(this)
}
is HttpContent -> ctx.writeAndFlush(msg)
else -> super.channelRead(ctx, msg)
}
}
override fun exceptionCaught(ctx: ChannelHandlerContext?, cause: Throwable?) {
super.exceptionCaught(ctx, cause)
}
}
@@ -0,0 +1,32 @@
package net.woggioni.rbcs.server.otel
import io.netty.channel.ChannelHandler
import io.opentelemetry.api.GlobalOpenTelemetry
import io.opentelemetry.instrumentation.netty.v4_1.NettyServerTelemetry
import net.woggioni.rbcs.common.createLogger
import net.woggioni.rbcs.common.warn
object OtelIntegration {
private val log = createLogger<OtelIntegration>()
val isAvailable: Boolean by lazy {
runCatching {
Class.forName("io.opentelemetry.api.OpenTelemetry")
}.fold(
onSuccess = { true },
onFailure = {
log.warn { "OpenTelemetry classes not on classpath, instrumentation disabled" }
false
},
)
}
fun createHandler(): ChannelHandler? {
return if (isAvailable) createHandlerInternal() else null
}
private fun createHandlerInternal(): ChannelHandler {
return NettyServerTelemetry.create(GlobalOpenTelemetry.get()).createCombinedHandler()
}
}
@@ -0,0 +1,8 @@
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<rbcs:server
xmlns:xs="http://www.w3.org/2001/XMLSchema-instance"
xmlns:rbcs="urn:net.woggioni.rbcs.server"
xs:schemaLocation="urn:net.woggioni.rbcs.server jpms://net.woggioni.rbcs.server/net/woggioni/rbcs/server/schema/rbcs-server.xsd">
<bind host="127.0.0.1" port="8080" incoming-connections-backlog-size="1024"/>
<cache xs:type="rbcs:fileSystemCacheType" path="${sys:java.io.tmpdir}/rbcs" max-age="P7D"/>
</rbcs:server>
@@ -0,0 +1,767 @@
<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<xs:schema targetNamespace="urn:net.woggioni.rbcs.server"
xmlns:xs="http://www.w3.org/2001/XMLSchema"
xmlns:rbcs="urn:net.woggioni.rbcs.server"
elementFormDefault="unqualified">
<xs:element name="server" type="rbcs:serverType">
<xs:annotation>
<xs:documentation>
Root element containing the server configuration
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:complexType name="serverType">
<xs:sequence minOccurs="0">
<xs:element name="bind" type="rbcs:bindType" maxOccurs="1"/>
<xs:element name="connection" type="rbcs:connectionType" minOccurs="0" maxOccurs="1"/>
<xs:element name="event-executor" type="rbcs:eventExecutorType" minOccurs="0" maxOccurs="1"/>
<xs:element name="rate-limiter" type="rbcs:rateLimiterType" minOccurs="0" maxOccurs="1"/>
<xs:element name="cache" type="rbcs:cacheType" maxOccurs="1">
<xs:annotation>
<xs:documentation>
Cache storage backend implementation to use, more implementations can be added through
the use of plugins
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="authorization" type="rbcs:authorizationType" minOccurs="0">
<xs:key name="userId">
<xs:selector xpath="users/user"/>
<xs:field xpath="@name"/>
</xs:key>
<xs:keyref name="userRef" refer="rbcs:userId">
<xs:selector xpath="groups/group/users/user"/>
<xs:field xpath="@ref"/>
</xs:keyref>
</xs:element>
<xs:element name="authentication" type="rbcs:authenticationType" minOccurs="0" maxOccurs="1">
<xs:annotation>
<xs:documentation>
Mechanism to use to assign a username to a specific client
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="tls" type="rbcs:tlsType" minOccurs="0" maxOccurs="1">
<xs:annotation>
<xs:documentation>
Use TLS to encrypt all the communications
</xs:documentation>
</xs:annotation>
</xs:element>
</xs:sequence>
<xs:attribute name="path" type="xs:string" use="optional">
<xs:annotation>
<xs:documentation>
URI path prefix, if your rbcs is hosted at "http://www.example.com"
and this parameter is set to "cache", then all the requests will need to be sent at
"http://www.example.com/cache/KEY", where "KEY" is the cache entry KEY
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="bindType">
<xs:sequence minOccurs="0">
<xs:element name="trusted-proxies" type="rbcs:trustedProxiesType"/>
</xs:sequence>
<xs:attribute name="host" type="xs:token" use="required">
<xs:annotation>
<xs:documentation>Server bind address</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="port" type="xs:unsignedShort" use="required">
<xs:annotation>
<xs:documentation>Server port number</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="proxy-protocol" type="xs:boolean" use="optional">
<xs:annotation>
<xs:documentation>Enable proxy protocol</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="incoming-connections-backlog-size" type="xs:unsignedInt" use="optional" default="1024">
<xs:annotation>
<xs:documentation>
The maximum queue length for incoming connection indications (a request to connect) is set to
the backlog parameter. If a connection indication arrives when the queue is full,
the connection is refused.
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="trustedProxiesType">
<xs:sequence minOccurs="0">
<xs:element name="allow" type="rbcs:allowType" maxOccurs="unbounded"/>
</xs:sequence>
</xs:complexType>
<xs:complexType name="allowType">
<xs:attribute name="cidr" type="rbcs:cidr"/>
</xs:complexType>
<xs:complexType name="connectionType">
<xs:attribute name="idle-timeout" type="xs:duration" use="optional" default="PT30S">
<xs:annotation>
<xs:documentation>
The server will close the connection with the client
when neither a read nor a write was performed for the specified period of time.
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="read-idle-timeout" type="xs:duration" use="optional" default="PT60S">
<xs:annotation>
<xs:documentation>
The server will close the connection with the client
when no read was performed for the specified period of time.
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="write-idle-timeout" type="xs:duration" use="optional" default="PT60S">
<xs:annotation>
<xs:documentation>
The server will close the connection with the client
when no write was performed for the specified period of time.
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="max-request-size" type="rbcs:byteSizeType" use="optional" default="0x4000000">
<xs:annotation>
<xs:documentation>
The maximum request body size the server will accept from a client
(if exceeded the server returns 413 HTTP status code)
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="chunk-size" type="rbcs:byteSizeType" default="0x10000">
<xs:annotation>
<xs:documentation>
Maximum byte size of socket write calls
(reduce it to reduce memory consumption, increase it for increased throughput)
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="eventExecutorType">
<xs:attribute name="use-virtual-threads" type="xs:boolean" use="optional" default="true">
<xs:annotation>
<xs:documentation>
Whether or not to use virtual threads for the execution of the core server handler
(not for the I/O operations)
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="rateLimiterType">
<xs:attribute name="delay-response" type="xs:boolean" use="optional" default="false">
<xs:annotation>
<xs:documentation>
If set to true, the server will delay responses to meet user quotas, otherwise it will simply
return an immediate 429 status code to all requests that exceed the configured quota
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="max-queued-messages" type="xs:nonNegativeInteger" use="optional" default="100">
<xs:annotation>
<xs:documentation>
Only meaningful when "delay-response" is set to "true",
when a request is delayed, it and all the following messages are queued
as long as "max-queued-messages" is not crossed, all requests that would exceed the
max-queued-message limit are instead discarded and responded with a 429 status code
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="message-buffer-size" type="rbcs:byteSizeType" use="optional" default="0x100000">
<xs:annotation>
<xs:documentation>
Only meaningful when "delay-response" is set to "true",
when a request is delayed, it and all the following requests are buffered
as long as "message-buffer-size" is not crossed, all requests that would exceed the buffer
size are instead discarded and responded with a 429 status code
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="cacheType" abstract="true"/>
<xs:complexType name="inMemoryCacheType">
<xs:annotation>
<xs:documentation>
A simple cache implementation that uses a java.util.ConcurrentHashMap as a storage backend
</xs:documentation>
</xs:annotation>
<xs:complexContent>
<xs:extension base="rbcs:cacheType">
<xs:attribute name="max-age" type="xs:duration" default="P1D">
<xs:annotation>
<xs:documentation>
Values will be removed from the cache after this amount of time
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="max-size" type="rbcs:byteSizeType" default="0x1000000">
<xs:annotation>
<xs:documentation>
The maximum allowed total size of the cache in bytes, old values will be purged from the cache
when the insertion of a new value causes this limit to be exceeded
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="digest" type="xs:token">
<xs:annotation>
<xs:documentation>
Hashing algorithm to apply to the key. If omitted, no hashing is performed.
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="enable-compression" type="xs:boolean" default="true">
<xs:annotation>
<xs:documentation>
Enable deflate compression for stored cache elements
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="compression-level" type="rbcs:compressionLevelType" default="-1">
<xs:annotation>
<xs:documentation>
Deflate compression level to use for cache compression,
use -1 to use the default compression level of java.util.zip.Deflater
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:extension>
</xs:complexContent>
</xs:complexType>
<xs:complexType name="fileSystemCacheType">
<xs:annotation>
<xs:documentation>
A simple cache implementation that stores data in a folder on the filesystem
</xs:documentation>
</xs:annotation>
<xs:complexContent>
<xs:extension base="rbcs:cacheType">
<xs:attribute name="path" type="xs:string" use="optional">
<xs:annotation>
<xs:documentation>
File system path that will be used to store the cache data files
(it will be created if it doesn't already exist)
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="max-age" type="xs:duration" default="P1D">
<xs:annotation>
<xs:documentation>
Values will be removed from the cache after this amount of time
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="digest" type="xs:token" default="SHA3-224">
<xs:annotation>
<xs:documentation>
Hashing algorithm to apply to the key. If omitted, no hashing is performed.
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="enable-compression" type="xs:boolean" default="true">
<xs:annotation>
<xs:documentation>
Enable deflate compression for stored cache elements
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="compression-level" type="rbcs:compressionLevelType" default="-1">
<xs:annotation>
<xs:documentation>
Deflate compression level to use for cache compression,
use -1 to use the default compression level of java.util.zip.Deflater
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:extension>
</xs:complexContent>
</xs:complexType>
<xs:complexType name="tlsCertificateAuthorizationType">
<xs:sequence>
<xs:element name="group-extractor" type="rbcs:X500NameExtractorType" minOccurs="0">
<xs:annotation>
<xs:documentation>
A regex based extractor that will be used to determine which group the client belongs to,
based on the X.500 name of the subject field in the client's TLS certificate.
When this is set RBAC works even if the user isn't listed in the &lt;users/&gt; section as
the client will be assigned role solely based on the group he is found to belong to.
Note that this does not allow for a client to be part of multiple groups.
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="user-extractor" type="rbcs:X500NameExtractorType" minOccurs="0">
<xs:annotation>
<xs:documentation>
A regex based extractor that will be used to assign a user to a connected client,
based on the X.500 name of the subject field in the client's TLS certificate.
</xs:documentation>
</xs:annotation>
</xs:element>
</xs:sequence>
</xs:complexType>
<xs:complexType name="forwardedClientCertificateAuthorizationType">
<xs:annotation>
<xs:documentation>
Authenticate clients based on a custom HTTP header containing the client TLS certificate
subject DN, forwarded by a reverse proxy that performs TLS termination. The proxy must be
listed in the trusted-proxies configuration for the header to be accepted.
</xs:documentation>
</xs:annotation>
<xs:sequence>
<xs:element name="group-extractor" type="rbcs:X500NameExtractorType" minOccurs="0">
<xs:annotation>
<xs:documentation>
A regex based extractor that will be used to determine which group the client belongs to,
based on the X.500 name of the subject DN forwarded by the reverse proxy.
When this is set RBAC works even if the user isn't listed in the &lt;users/&gt; section as
the client will be assigned role solely based on the group he is found to belong to.
Note that this does not allow for a client to be part of multiple groups.
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="user-extractor" type="rbcs:X500NameExtractorType" minOccurs="0">
<xs:annotation>
<xs:documentation>
A regex based extractor that will be used to assign a user to a connected client,
based on the X.500 name of the subject DN forwarded by the reverse proxy.
</xs:documentation>
</xs:annotation>
</xs:element>
</xs:sequence>
<xs:attribute name="header-name" type="xs:token">
<xs:annotation>
<xs:documentation>
Name of the HTTP header containing the client certificate subject DN
forwarded by the reverse proxy. Defaults to "X-Client-Cert-Subject-DN".
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="X500NameExtractorType">
<xs:annotation>
<xs:documentation>
Extract informations from a client TLS certificates using
regular expressions applied to the X.500 name "Subject" field
</xs:documentation>
</xs:annotation>
<xs:attribute name="attribute-name" type="xs:token">
<xs:annotation>
<xs:documentation>
X.500 name attribute to apply the regex
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="pattern" type="xs:token">
<xs:annotation>
<xs:documentation>
Regex that wil be applied to the attribute value,
use regex groups to extract relevant data
(note that only the first group that appears in the regex is used)
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="authorizationType">
<xs:all>
<xs:element name="users" type="rbcs:usersType">
<xs:annotation>
<xs:documentation>
List of users registered in the application
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="groups" type="rbcs:groupsType">
<xs:annotation>
<xs:documentation>
List of user groups registered in the application
</xs:documentation>
</xs:annotation>
<xs:unique name="groupKey">
<xs:selector xpath="group"/>
<xs:field xpath="@name"/>
</xs:unique>
</xs:element>
</xs:all>
</xs:complexType>
<xs:complexType name="authenticationType">
<xs:annotation>
<xs:documentation>
Authentication mechanism to assign usernames and groups to clients
</xs:documentation>
</xs:annotation>
<xs:choice>
<xs:element name="basic">
<xs:annotation>
<xs:documentation>
Enable HTTP basic authentication
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="client-certificate" type="rbcs:tlsCertificateAuthorizationType">
<xs:annotation>
<xs:documentation>
Enable TLS certificate authentication
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="forwarded-client-certificate" type="rbcs:forwardedClientCertificateAuthorizationType">
<xs:annotation>
<xs:documentation>
Enable forwarded client certificate authentication. Authenticates clients based on
a custom HTTP header containing the client certificate subject DN, forwarded by a
reverse proxy that performs TLS termination. Requires trusted-proxies to be configured.
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="none">
<xs:annotation>
<xs:documentation>
Disable authentication altogether
</xs:documentation>
</xs:annotation>
</xs:element>
</xs:choice>
</xs:complexType>
<xs:complexType name="quotaType">
<xs:annotation>
<xs:documentation>
Defines a quota for a user or a group
</xs:documentation>
</xs:annotation>
<xs:attribute name="calls" type="xs:positiveInteger" use="required">
<xs:annotation>
<xs:documentation>
Maximum number of allowed calls in a given period
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="period" type="xs:duration" use="required">
<xs:annotation>
<xs:documentation>
The period length
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="max-available-calls" type="xs:positiveInteger" use="optional">
<xs:annotation>
<xs:documentation>
Maximum number of available calls that can be accumulated
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="initial-available-calls" type="xs:unsignedInt" use="optional">
<xs:annotation>
<xs:documentation>
Number of available calls for users at their first call
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="anonymousUserType">
<xs:annotation>
<xs:documentation>
Placeholder for a client that is not authenticated
</xs:documentation>
</xs:annotation>
<xs:sequence>
<xs:element name="quota" type="rbcs:quotaType" minOccurs="0" maxOccurs="1">
<xs:annotation>
<xs:documentation>
Calls quota for the user
</xs:documentation>
</xs:annotation>
</xs:element>
</xs:sequence>
</xs:complexType>
<xs:complexType name="userType">
<xs:annotation>
<xs:documentation>
An authenticated user
</xs:documentation>
</xs:annotation>
<xs:sequence>
<xs:element name="quota" type="rbcs:quotaType" minOccurs="0" maxOccurs="1">
<xs:annotation>
<xs:documentation>
Calls quota for the user
</xs:documentation>
</xs:annotation>
</xs:element>
</xs:sequence>
<xs:attribute name="name" type="xs:token" use="required">
<xs:annotation>
<xs:documentation>
User's name
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="password" type="xs:string" use="optional">
<xs:annotation>
<xs:documentation>
User's password hash used for HTTP basic authentication, this has to be generated with
the `password` subcommand of `rbcs-cli`
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="usersType">
<xs:annotation>
<xs:documentation>
List of registered users, add an &lt;anonymous&gt; tag to enable authenticated user access
when authentication is enabled
</xs:documentation>
</xs:annotation>
<xs:sequence>
<xs:element name="user" type="rbcs:userType" minOccurs="0" maxOccurs="unbounded"/>
<xs:element name="anonymous" type="rbcs:anonymousUserType" minOccurs="0" maxOccurs="1"/>
</xs:sequence>
</xs:complexType>
<xs:complexType name="groupsType">
<xs:annotation>
<xs:documentation>
List of registered user groups
</xs:documentation>
</xs:annotation>
<xs:sequence>
<xs:element name="group" type="rbcs:groupType" maxOccurs="unbounded" minOccurs="0"/>
</xs:sequence>
</xs:complexType>
<xs:complexType name="groupType">
<xs:annotation>
<xs:documentation>
The definition of a user group, with the list of its member users
</xs:documentation>
</xs:annotation>
<xs:sequence>
<xs:element name="users" type="rbcs:userRefsType" maxOccurs="1" minOccurs="0">
<xs:unique name="userRefWriterKey">
<xs:selector xpath="user"/>
<xs:field xpath="@ref"/>
</xs:unique>
</xs:element>
<xs:element name="roles" type="rbcs:rolesType" maxOccurs="1" minOccurs="0">
<xs:annotation>
<xs:documentation>
The list of application roles awarded to all the members of this group
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="user-quota" type="rbcs:quotaType" minOccurs="0" maxOccurs="1">
<xs:annotation>
<xs:documentation>
The call quota for each user in this group
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="group-quota" type="rbcs:quotaType" minOccurs="0" maxOccurs="1">
<xs:annotation>
<xs:documentation>
The cumulative call quota for all users in this group
</xs:documentation>
</xs:annotation>
</xs:element>
</xs:sequence>
<xs:attribute name="name" type="xs:token">
<xs:annotation>
<xs:documentation>
The group's name
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="rolesType">
<xs:sequence>
<xs:choice maxOccurs="unbounded">
<xs:element name="writer"/>
<xs:element name="reader"/>
<xs:element name="healthcheck"/>
</xs:choice>
</xs:sequence>
</xs:complexType>
<xs:complexType name="userRefsType">
<xs:annotation>
<xs:documentation>
A list of references to users in the &lt;users&gt; section
</xs:documentation>
</xs:annotation>
<xs:sequence>
<xs:element name="user" type="rbcs:userRefType" maxOccurs="unbounded" minOccurs="0"/>
<xs:element name="anonymous" minOccurs="0" maxOccurs="1"/>
</xs:sequence>
</xs:complexType>
<xs:complexType name="userRefType">
<xs:annotation>
<xs:documentation>
A reference to a user in the &lt;users&gt; section
</xs:documentation>
</xs:annotation>
<xs:attribute name="ref" type="xs:string" use="required">
<xs:annotation>
<xs:documentation>
Name of the referenced user
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="tlsType">
<xs:annotation>
<xs:documentation>
Enable TLS protocol
</xs:documentation>
</xs:annotation>
<xs:all>
<xs:element name="keystore" type="rbcs:keyStoreType" >
<xs:annotation>
<xs:documentation>
Path to the keystore file that contains the server's key and certificate
</xs:documentation>
</xs:annotation>
</xs:element>
<xs:element name="truststore" type="rbcs:trustStoreType" minOccurs="0">
<xs:annotation>
<xs:documentation>
Path to the truststore file that contains the trusted CAs
for TLS client certificate verification
</xs:documentation>
</xs:annotation>
</xs:element>
</xs:all>
</xs:complexType>
<xs:complexType name="keyStoreType">
<xs:attribute name="file" type="xs:string" use="required">
<xs:annotation>
<xs:documentation>
System path to the keystore file
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="password" type="xs:string">
<xs:annotation>
<xs:documentation>
Password to open they keystore file
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="key-alias" type="xs:string" use="required">
<xs:annotation>
<xs:documentation>
Alias of the keystore entry containing the private key
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="key-password" type="xs:string">
<xs:annotation>
<xs:documentation>
Private key entry's encryption password
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="trustStoreType">
<xs:attribute name="file" type="xs:string" use="required">
<xs:annotation>
<xs:documentation>
Path to the trustore file
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="password" type="xs:string">
<xs:annotation>
<xs:documentation>
Trustore file password
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="check-certificate-status" type="xs:boolean">
<xs:annotation>
<xs:documentation>
Whether or not check the certificate validity using CRL/OCSP
</xs:documentation>
</xs:annotation>
</xs:attribute>
<xs:attribute name="require-client-certificate" type="xs:boolean" use="optional" default="false">
<xs:annotation>
<xs:documentation>
If true, the server requires a TLS client certificate from the client and simply refuses to connect
when a client certificate isn't provided
</xs:documentation>
</xs:annotation>
</xs:attribute>
</xs:complexType>
<xs:complexType name="propertiesType">
<xs:sequence>
<xs:element maxOccurs="unbounded" minOccurs="0" name="property" type="rbcs:propertyType"/>
</xs:sequence>
</xs:complexType>
<xs:complexType name="propertyType">
<xs:simpleContent>
<xs:extension base="xs:string">
<xs:attribute name="key" type="xs:string" use="required"/>
</xs:extension>
</xs:simpleContent>
</xs:complexType>
<xs:complexType name="hostAndPortType">
<xs:attribute name="host" type="xs:string" use="required"/>
<xs:attribute name="port" type="xs:unsignedShort" use="required"/>
</xs:complexType>
<xs:simpleType name="byteSizeType">
<xs:restriction base="xs:token">
<xs:pattern value="(0x[a-f0-9]+|[0-9]+)"/>
</xs:restriction>
</xs:simpleType>
<xs:simpleType name="compressionLevelType">
<xs:restriction base="xs:integer">
<xs:minInclusive value="-1"/>
<xs:maxInclusive value="9"/>
</xs:restriction>
</xs:simpleType>
<xs:simpleType name="cidrIPv4">
<xs:restriction base="xs:string">
<xs:pattern value="(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\/(3[0-2]|[12]?[0-9])" />
</xs:restriction>
</xs:simpleType>
<xs:simpleType name="cidrIPv6">
<xs:restriction base="xs:string">
<xs:pattern value="((([0-9A-Fa-f]{1,4}:){7}([0-9A-Fa-f]{1,4}|:))|(([0-9A-Fa-f]{1,4}:){6}(:[0-9A-Fa-f]{1,4}|((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){5}(((:[0-9A-Fa-f]{1,4}){1,2})|:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3})|:))|(([0-9A-Fa-f]{1,4}:){4}(((:[0-9A-Fa-f]{1,4}){1,3})|((:[0-9A-Fa-f]{1,4})?:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){3}(((:[0-9A-Fa-f]{1,4}){1,4})|((:[0-9A-Fa-f]{1,4}){0,2}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){2}(((:[0-9A-Fa-f]{1,4}){1,5})|((:[0-9A-Fa-f]{1,4}){0,3}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(([0-9A-Fa-f]{1,4}:){1}(((:[0-9A-Fa-f]{1,4}){1,6})|((:[0-9A-Fa-f]{1,4}){0,4}:((25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)(\.(25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)){3}))|:))|(:(:([0-9A-Fa-f]{1,4}){1,7}|:)))(%[\p{L}\p{N}_-]+)?\/(12[0-8]|1[01][0-9]|[1-9]?[0-9])" />
</xs:restriction>
</xs:simpleType>
<xs:simpleType name="cidr">
<xs:union memberTypes="rbcs:cidrIPv4 rbcs:cidrIPv6" />
</xs:simpleType>
</xs:schema>
@@ -0,0 +1,91 @@
package net.woggioni.rbcs.server.throttling
import java.net.InetSocketAddress
import java.util.Arrays
import java.util.concurrent.ConcurrentHashMap
import java.util.function.Function
import net.woggioni.jwo.Bucket
import net.woggioni.rbcs.api.Configuration
class BucketManager private constructor(
private val bucketsByUser: Map<Configuration.User, List<Bucket>> = HashMap(),
private val bucketsByGroup: Map<Configuration.Group, Bucket> = HashMap(),
loader: Function<InetSocketAddress, Bucket>?
) {
private class BucketsByAddress(
private val map: MutableMap<ByteArrayKey, Bucket>,
private val loader: Function<InetSocketAddress, Bucket>
) {
fun getBucket(socketAddress : InetSocketAddress) = map.computeIfAbsent(ByteArrayKey(socketAddress.address.address)) {
loader.apply(socketAddress)
}
}
private val bucketsByAddress: BucketsByAddress? = loader?.let {
BucketsByAddress(ConcurrentHashMap(), it)
}
private class ByteArrayKey(val array: ByteArray) {
override fun equals(other: Any?) = (other as? ByteArrayKey)?.let { bak ->
array contentEquals bak.array
} ?: false
override fun hashCode() = Arrays.hashCode(array)
}
fun getBucketByAddress(address : InetSocketAddress) : Bucket? {
return bucketsByAddress?.getBucket(address)
}
fun getBucketByUser(user : Configuration.User) = bucketsByUser[user]
fun getBucketByGroup(group : Configuration.Group) = bucketsByGroup[group]
companion object {
fun from(cfg : Configuration) : BucketManager {
val bucketsByUser = cfg.users.values.asSequence().map { user ->
val buckets = (
user.quota
?.let { quota ->
sequenceOf(quota)
} ?: user.groups.asSequence()
.mapNotNull(Configuration.Group::getUserQuota)
).map { quota ->
Bucket.local(
quota.maxAvailableCalls,
quota.calls,
quota.period,
quota.initialAvailableCalls
)
}.toList()
user to buckets
}.toMap()
val bucketsByGroup = cfg.groups.values.asSequence().filter {
it.groupQuota != null
}.map { group ->
val quota = group.groupQuota
val bucket = Bucket.local(
quota.maxAvailableCalls,
quota.calls,
quota.period,
quota.initialAvailableCalls
)
group to bucket
}.toMap()
return BucketManager(
bucketsByUser,
bucketsByGroup,
cfg.users[""]?.quota?.let { anonymousUserQuota ->
Function {
Bucket.local(
anonymousUserQuota.maxAvailableCalls,
anonymousUserQuota.calls,
anonymousUserQuota.period,
anonymousUserQuota.initialAvailableCalls
)
}
}
)
}
}
}
@@ -0,0 +1,228 @@
package net.woggioni.rbcs.server.throttling
import io.netty.buffer.ByteBufHolder
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInboundHandlerAdapter
import io.netty.handler.codec.http.DefaultFullHttpResponse
import io.netty.handler.codec.http.FullHttpMessage
import io.netty.handler.codec.http.HttpHeaderNames
import io.netty.handler.codec.http.HttpRequest
import io.netty.handler.codec.http.HttpResponseStatus
import io.netty.handler.codec.http.HttpVersion
import io.netty.handler.codec.http.LastHttpContent
import java.net.InetSocketAddress
import java.time.Duration
import java.time.temporal.ChronoUnit
import java.util.ArrayDeque
import java.util.concurrent.TimeUnit
import net.woggioni.jwo.Bucket
import net.woggioni.jwo.LongMath
import net.woggioni.rbcs.api.Configuration
import net.woggioni.rbcs.common.createLogger
import net.woggioni.rbcs.common.debug
import net.woggioni.rbcs.server.RemoteBuildCacheServer
class ThrottlingHandler(
private val bucketManager: BucketManager,
rateLimiterConfiguration: Configuration.RateLimiter,
connectionConfiguration: Configuration.Connection
) : ChannelInboundHandlerAdapter() {
private companion object {
private val log = createLogger<ThrottlingHandler>()
fun nextAttemptIsWithinThreshold(nextAttemptNanos : Long, waitThreshold : Duration) : Boolean {
val waitDuration = Duration.of(LongMath.ceilDiv(nextAttemptNanos, 100_000_000L) * 100L, ChronoUnit.MILLIS)
return waitDuration < waitThreshold
}
}
private class RefusedRequest
private val maxMessageBufferSize = rateLimiterConfiguration.messageBufferSize
private val maxQueuedMessages = rateLimiterConfiguration.maxQueuedMessages
private val delayRequests = rateLimiterConfiguration.isDelayRequest
private var requestBufferSize : Int = 0
private var valveClosed = false
private var queuedContent = ArrayDeque<Any>()
/**
* If the suggested waiting time from the bucket is lower than this
* amount, then the server will simply wait by itself before sending a response
* instead of replying with 429
*/
private val waitThreshold = minOf(
connectionConfiguration.idleTimeout,
connectionConfiguration.readIdleTimeout,
connectionConfiguration.writeIdleTimeout
).dividedBy(2)
override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
if(valveClosed) {
if(msg !is HttpRequest && msg is ByteBufHolder) {
val newBufferSize = requestBufferSize + msg.content().readableBytes()
if(newBufferSize > maxMessageBufferSize || queuedContent.size + 1 > maxQueuedMessages) {
log.debug {
if (newBufferSize > maxMessageBufferSize) {
"New message part exceeds maxMessageBufferSize, removing previous chunks"
} else {
"New message part exceeds maxQueuedMessages, removing previous chunks"
}
}
// If this message overflows the maxMessageBufferSize,
// then remove the previously enqueued chunks of the request from the deque,
// then discard the message
while(true) {
val tail = queuedContent.last()
if(tail is ByteBufHolder) {
requestBufferSize -= tail.content().readableBytes()
tail.release()
}
queuedContent.removeLast()
if(tail is HttpRequest) {
break
}
}
msg.release()
//Add a placeholder to remember to return a 429 response corresponding to this request
queuedContent.addLast(RefusedRequest())
} else {
//If the message does not overflow maxMessageBufferSize, just add it to the deque
queuedContent.addLast(msg)
requestBufferSize = newBufferSize
}
} else if(msg is HttpRequest && msg is FullHttpMessage){
val newBufferSize = requestBufferSize + msg.content().readableBytes()
// If this message overflows the maxMessageBufferSize,
// discard the message
if(newBufferSize > maxMessageBufferSize || queuedContent.size + 1 > maxQueuedMessages) {
log.debug {
if (newBufferSize > maxMessageBufferSize) {
"New message exceeds maxMessageBufferSize, discarding it"
} else {
"New message exceeds maxQueuedMessages, discarding it"
}
}
msg.release()
//Add a placeholder to remember to return a 429 response corresponding to this request
queuedContent.addLast(RefusedRequest())
} else {
//If the message does not exceed maxMessageBufferSize or maxQueuedMessages, just add it to the deque
queuedContent.addLast(msg)
requestBufferSize = newBufferSize
}
} else {
queuedContent.addLast(msg)
}
} else {
entryPoint(ctx, msg)
}
}
private fun entryPoint(ctx : ChannelHandlerContext, msg : Any) {
if(msg is RefusedRequest) {
sendThrottledResponse(ctx, null)
if(queuedContent.isEmpty()) {
valveClosed = false
} else {
val head = queuedContent.poll()
if(head is ByteBufHolder) {
requestBufferSize -= head.content().readableBytes()
}
entryPoint(ctx, head)
}
} else if(msg is HttpRequest) {
val nextAttempt = getNextAttempt(ctx)
if (nextAttempt < 0) {
super.channelRead(ctx, msg)
if(msg !is LastHttpContent) {
while (true) {
val head = queuedContent.poll() ?: break
if(head is ByteBufHolder) {
requestBufferSize -= head.content().readableBytes()
}
super.channelRead(ctx, head)
if (head is LastHttpContent) break
}
}
if(queuedContent.isEmpty()) {
valveClosed = false
} else {
val head = queuedContent.poll()
if(head is ByteBufHolder) {
requestBufferSize -= head.content().readableBytes()
}
entryPoint(ctx, head)
}
} else {
val waitDuration = Duration.of(LongMath.ceilDiv(nextAttempt, 100_000_000L) * 100L, ChronoUnit.MILLIS)
if (delayRequests && nextAttemptIsWithinThreshold(nextAttempt, waitThreshold)) {
valveClosed = true
ctx.executor().schedule({
entryPoint(ctx, msg)
}, waitDuration.toMillis(), TimeUnit.MILLISECONDS)
} else {
sendThrottledResponse(ctx, waitDuration)
if(queuedContent.isEmpty()) {
valveClosed = false
} else {
val head = queuedContent.poll()
if(head is ByteBufHolder) {
requestBufferSize -= head.content().readableBytes()
}
entryPoint(ctx, head)
}
}
}
} else {
super.channelRead(ctx, msg)
}
}
/**
* Returns the number amount of milliseconds to wait before the requests can be processed
* or -1 if the request can be performed immediately
*/
private fun getNextAttempt(ctx : ChannelHandlerContext) : Long {
val buckets = mutableListOf<Bucket>()
val user = ctx.channel().attr(RemoteBuildCacheServer.userAttribute).get()
if (user != null) {
bucketManager.getBucketByUser(user)?.let(buckets::addAll)
}
val groups = ctx.channel().attr(RemoteBuildCacheServer.groupAttribute).get() ?: emptySet()
if (groups.isNotEmpty()) {
groups.forEach { group ->
bucketManager.getBucketByGroup(group)?.let(buckets::add)
}
}
if (user == null && groups.isEmpty()) {
val clientAddress = ctx.channel().attr<InetSocketAddress>(RemoteBuildCacheServer.clientIp).get()
bucketManager.getBucketByAddress(clientAddress)?.let(buckets::add)
}
var nextAttempt = -1L
for (bucket in buckets) {
val bucketNextAttempt = bucket.removeTokensWithEstimate(1)
if (bucketNextAttempt > nextAttempt) {
nextAttempt = bucketNextAttempt
}
}
return nextAttempt
}
private fun sendThrottledResponse(ctx: ChannelHandlerContext, retryAfter: Duration?) {
val response = DefaultFullHttpResponse(
HttpVersion.HTTP_1_1,
HttpResponseStatus.TOO_MANY_REQUESTS
)
response.headers()[HttpHeaderNames.CONTENT_LENGTH] = 0
retryAfter?.seconds?.takeIf {
it > 0
}?.let {
response.headers()[HttpHeaderNames.RETRY_AFTER] = it
}
ctx.writeAndFlush(response)
}
}