Support docker and swarm.

README.md

@@ -6,10 +6,19 @@
 
 Passport 是一个超轻量级统一认证网关, 面向使用 [钉钉](https://www.dingtalk.com) 或是 [企业微信](https://work.weixin.qq.com/) 的创业团队提供手机扫码登录访问内部服务.
 
+## 跑起来
+
+```sh
+curl -LkO https://github.com/zhongl/passport/raw/master/docker-compose.yml 
+curl -LkO https://github.com/zhongl/passport/raw/master/app.conf
+DOMAIN=foo.bar docker-compose up -d
+curl -k -v https://localhost -H 'Host: www.foo.bar' -H 'Cookie: jwt=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJwYXNzcG9ydCIsIm5hbWUiOiJ6aG9uZ2wiLCJleHAiOjE4NjYxNzI3MjV9.FomLr4SgRvHuI6iUnVZc2-Q9YQbNrh4eDWGbM09xoC8'
+```
+
+- 细节请详见 [docker-compose.yml](https://github.com/zhongl/passport/blob/master/docker-compose.yml) . 
 
-# 部署
 
-## 配置 app.conf
+## 配置 
 
 ### 钉钉
 
@@ -62,12 +71,6 @@ wechat {
 
 > 参见[企业内部开发](https://work.weixin.qq.com/api/doc#90000/90003/90487), 创建**应用**.
 
-## 运行
-
-```sh
-docker run -d -v $(pwd)/app.conf:/app.conf -e JAVA_OPTS=-Dconfig.file=/app.conf zhongl/passport
-```
-
 ## Echo调试
 
 若需要在真正部署之前进行调试验证, 可在运行时指定`-e`:

docker-compose.yml

@@ -1,47 +1,44 @@
 version: '3.3'
 services:
 
-#  gateway:
-#    image: traefik:latest
-#    command:
-#      - "--api"
-#      - "--entrypoints=Name:http Address::80 Redirect.EntryPoint:https"
-#      - "--entrypoints=Name:https Address::443 TLS"
-#      - "--defaultentrypoints=http,https"
-#      - "--docker"
-#      - "--docker.watch"
-#      - "--docker.domain=${DOMAIN}"
-#      - "--accesslog"
-#      - "--traefikLog"
-#    volumes:
-#      - /var/run/docker.sock:/var/run/docker.sock
-#    ports:
-#      - 80:80
-#      - 443:443
+  gateway:
+    image: traefik:latest
+    command:
+      - "--api"
+      - "--entrypoints=Name:http Address::80 Redirect.EntryPoint:https"
+      - "--entrypoints=Name:https Address::443 TLS"
+      - "--defaultentrypoints=http,https"
+      - "--docker"
+      - "--docker.watch"
+      - "--docker.domain=${DOMAIN}"
+      - "--accesslog"
+      - "--traefikLog"
+    volumes:
+      - /var/run/docker.sock:/var/run/docker.sock
+    ports:
+      - 80:80
+      - 443:443
 
   passport:
     image: zhongl/passport:latest
-    command:
-      - "-d:docker"
     volumes:
       - /var/run/docker.sock:/var/run/docker.sock
-#    labels:
-#      traefik.port: 8080
-#      traefik.frontend.rule: "HostRegexp: {subdomain:[a-z]+}.${DOMAIN}"
+    labels:
+      traefik.port: 8080
+      traefik.frontend.rule: "HostRegexp: {subdomain:[a-z]+}.${DOMAIN}"
     environment:
       JAVA_TOOL_OPTIONS: -Dconfig.file=/run/secrets/conf
       DOMAIN: $DOMAIN
     secrets:
       - conf
-    ports:
-      - 80:8080
 
   echo:
     image: zhongl/passport:latest
     command:
       - "-e"
     labels:
-      passport.rule: "www.${DOMAIN}|>|:8080"
+      passport.rule: "www.${DOMAIN}"
+      passport.port: 8080
 
 secrets:
   conf:

src/main/scala/zhongl/passport/CommandLine.scala

@@ -19,7 +19,7 @@ package zhongl.passport
 import scopt.OptionParser
 
 object CommandLine {
-  case class Opt(host: String = "0.0.0.0", port: Int = 8080, echo: Boolean = false, dynamic: String = "docker")
+  case class Opt(host: String = "0.0.0.0", port: Int = 8080, echo: Boolean = false, dynamic: Docker.Mode.Value = Docker.Mode.Local)
 
   val parser = new OptionParser[Opt]("passport") {
     head("passport", "0.0.1")
@@ -37,8 +37,8 @@ object CommandLine {
       .text("enable echo mode for debug, default is disable")
 
     arg[String]("<dynamic mode>")
-      .required()
-      .action((x, c) => c.copy(dynamic = x))
+      .action { case (Docker.Mode(value), c) => c.copy(dynamic = value) }
+      .optional()
       .validate {
         case "docker" | "swarm" => success
         case _                  => failure("dynamic mode must be docker or swarm.")

src/main/scala/zhongl/passport/Docker.scala

@@ -17,19 +17,17 @@
 package zhongl.passport
 
 import java.io.File
-import java.net.InetSocketAddress
 
 import akka.NotUsed
 import akka.actor.ActorSystem
 import akka.event.Logging
+import akka.http.scaladsl.Http
 import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport
 import akka.http.scaladsl.model.HttpEntity.{ChunkStreamPart, Chunked}
 import akka.http.scaladsl.model.Uri.Path
 import akka.http.scaladsl.model._
 import akka.http.scaladsl.model.headers.Host
-import akka.http.scaladsl.settings.ClientConnectionSettings
 import akka.http.scaladsl.unmarshalling.{Unmarshal, Unmarshaller}
-import akka.http.scaladsl.{ClientTransport, Http}
 import akka.stream.alpakka.unixdomainsocket.scaladsl.UnixDomainSocket
 import akka.stream.scaladsl._
 import akka.stream.{ActorMaterializer, Attributes, FlowShape, Graph}
@@ -38,11 +36,11 @@ import spray.json._
 import zhongl.passport.Docker.{Container, Service}
 
 import scala.concurrent.Future
+import scala.util.matching.Regex
 
 class Docker(base: Uri, outgoing: () => Graph[FlowShape[HttpRequest, HttpResponse], Any])(implicit system: ActorSystem) extends Docker.JsonSupport {
 
   private implicit val mat = ActorMaterializer()
-  private implicit val ex  = system.dispatcher
 
   def events(filters: Map[String, List[String]]): Source[ByteString, Any] = {
     val query   = Uri.Query("filters" -> filters.toJson.compactPrint)
@@ -113,31 +111,57 @@ object Docker {
   final case class Service(`ID`: String, `Spec`: Spec)
   final case class Spec(`Name`: String, `Labels`: Map[String, String])
 
-  final class UnixSocketTransport(file: File) extends ClientTransport {
-    override def connectTo(
-        host: String,
-        port: Int,
-        settings: ClientConnectionSettings
-    )(implicit system: ActorSystem): Flow[ByteString, ByteString, Future[Http.OutgoingConnection]] = {
+  trait JsonSupport extends DefaultJsonProtocol with SprayJsonSupport {
+    implicit val specF: JsonFormat[Spec]           = jsonFormat2(Spec)
+    implicit val containerF: JsonFormat[Container] = jsonFormat3(Container)
+    implicit val serviceF: JsonFormat[Service]     = jsonFormat2(Service)
+  }
+
+  object Mode {
 
-      implicit val ex = system.dispatcher
-      val address     = InetSocketAddress.createUnresolved(host, port)
+    type Rules   = List[(Regex, Host)]
+    type Filters = Map[String, List[String]]
 
-      system.log.error(new Exception(), "connect")
+    private val rule = "passport.rule"
 
-      UnixDomainSocket()
-        .outgoingConnection(file)
-        .mapMaterializedValue(_.map { _ =>
-          system.log.error(new Exception(), "materialize")
-          Http.OutgoingConnection(address, address)
-        })
+    def unapply(arg: String): Option[Value] = arg match {
+      case "docker" => Some(Local)
+      case "swarm"  => Some(Swarm)
+      case _        => None
+    }
+
+    sealed trait Value {
+      def apply(docker: Docker): Source[Rules, NotUsed]
+    }
+
+    case object Local extends Value {
+      override def apply(docker: Docker): Source[Rules, NotUsed] = {
+        val filters = Map("scope" -> List("local"), "type" -> List("container"), "event" -> List("start", "destroy"))
+        val update = docker
+          .containers(_: Filters)
+          .map(_.map(c => c.`Labels`(rule).r -> Host(c.`Names`.head.substring(1), port(c.`Labels`))))
+        source(docker.events(filters), update)
+      }
+    }
+
+    case object Swarm extends Value {
+      override def apply(docker: Docker): Source[Rules, NotUsed] = {
+        val filters = Map("scope" -> List("swarm"), "type" -> List("service"), "event" -> List("update", "remove"))
+        val update = docker
+          .services(_: Filters)
+          .map(_.map(c => c.`Spec`.`Labels`(rule).r -> Host(c.`Spec`.`Name`, port(c.`Spec`.`Labels`))))
+        source(docker.events(filters), update)
+      }
+    }
+
+    private def port(labels: Map[String, String]): Int = {
+      labels.get("passport.port").map(_.toInt).getOrElse(0)
+    }
+
+    private def source(events: Source[ByteString, Any], update: Filters => Flow[Any, Rules, NotUsed]) = {
+      Source.single(ByteString.empty).concat(events).via(update(Map("label" -> List(rule))))
     }
-  }
 
-  trait JsonSupport extends DefaultJsonProtocol with SprayJsonSupport {
-    implicit val specF: JsonFormat[Spec]           = jsonFormat2(Spec)
-    implicit val containerF: JsonFormat[Container] = jsonFormat3(Container)
-    implicit val serviceF: JsonFormat[Service]     = jsonFormat2(Service)
   }
 
 }

src/main/scala/zhongl/passport/Handle.scala

@@ -22,7 +22,7 @@ import akka.http.scaladsl.model.headers.`Timeout-Access`
 import akka.http.scaladsl.model.{HttpRequest, HttpResponse, StatusCodes}
 import akka.http.scaladsl.util.FastFuture
 import akka.stream.FlowShape
-import akka.stream.scaladsl.{Flow, GraphDSL, Merge}
+import akka.stream.scaladsl.{Flow, GraphDSL, Merge, Source}
 import akka.util.Timeout
 import zhongl.stream.oauth2.Guard
 
@@ -33,7 +33,7 @@ import scala.util.control.NonFatal
 
 object Handle {
 
-  def apply(dynamic: String)(implicit system: ActorSystem): Flow[HttpRequest, HttpResponse, NotUsed] = {
+  def apply(dynamic: Source[Docker.Mode.Rules, Any])(implicit system: ActorSystem): Flow[HttpRequest, HttpResponse, NotUsed] = {
     val graph = GraphDSL.create() { implicit b =>
       import GraphDSL.Implicits._
 
@@ -68,7 +68,7 @@ object Handle {
     Guard.graph(plugin.oauth2(jc.generate), ignore)(system.dispatcher)
   }
 
-  private def rewriteShape(dynamic: String)(implicit system: ActorSystem) = {
+  private def rewriteShape(dynamic: Source[Docker.Mode.Rules, Any])(implicit system: ActorSystem) = {
     import Rewrite._
 
     implicit val timeout = Timeout(2.seconds)
@@ -80,7 +80,7 @@ object Handle {
 
     val base = IgnoreHeader(_.isInstanceOf[`Timeout-Access`]) & XForwardedFor(local)
 
-    val ref = system.actorOf(RewriteRequestActor.props(base, Docker())(dynamic), "RewriteRequest")
+    val ref = system.actorOf(RewriteRequestActor.props(dynamic, Some(base)), "RewriteRequest")
     Flow[HttpRequest].ask[Either[HttpResponse, HttpRequest]](ref)
   }
 }

src/main/scala/zhongl/passport/Main.scala

@@ -50,7 +50,7 @@ object Main extends Directives {
       case Opt(host, port, true, _) =>
         (host, port, Echo())
       case Opt(host, port, _, d) =>
-        (host, port, Handle(d))
+        (host, port, Handle(d(Docker())))
     } map {
       case (host, port, flow) => bind(flow, host, port)
     } getOrElse system.terminate()