Challenge / block session command from Kafka

banjax-config.yaml

@@ -11,7 +11,9 @@ global_decision_lists:
 iptables_ban_seconds: 10
 iptables_unbanner_seconds: 5
 kafka_brokers:
-  - "localhost:9092"
+  - kafkadev0.prod.deflect.network:9094
+  - kafkadev1.prod.deflect.network:9094
+  - kafkadev2.prod.deflect.network:9094
 kafka_security_protocol: 'ssl'
 kafka_ssl_ca: "/etc/banjax/caroot.pem"
 kafka_ssl_cert: "/etc/banjax/certificate.pem"
@@ -85,11 +87,11 @@ regexes_with_rates:
 sitewide_sha_inv_list:
   example.com: block
   foobar.com: no_block
-  localhost: no_block
+  sub.localhost: no_block
   www.localhost: no_block
 server_log_file: /var/log/banjax/banjax-format.log
 banning_log_file: /etc/banjax/ban_ip_list.log
-expiring_decision_ttl_seconds: 10
+expiring_decision_ttl_seconds: 100
 too_many_failed_challenges_interval_seconds: 10
 too_many_failed_challenges_threshold: 3
 password_cookie_ttl_seconds: 345600  # Dynamic apply to internal/password-protected-path.html:170
@@ -108,6 +110,6 @@ banning_log_file_temp: /etc/banjax/ban_ip_list_temp.log
 session_cookie_hmac_secret: some_secret
 session_cookie_ttl_seconds: 3600
 sites_to_disable_baskerville:
-  localhost: true
+  sub.localhost: false
 use_user_agent_in_cookie:
-  localhost: true
+  sub.localhost: true

internal/config.go

@@ -66,6 +66,7 @@ type Config struct {
 	DisableKafka                           bool              `yaml:"disable_kafka"`
 	SessionCookieHmacSecret                string            `yaml:"session_cookie_hmac_secret"`
 	SessionCookieTtlSeconds                int               `yaml:"session_cookie_ttl_seconds"`
+	SessionCookieNotVerify                 bool              `yaml:"session_cookie_not_verify"`
 	SitesToDisableBaskerville              map[string]bool   `yaml:"sites_to_disable_baskerville"`
 }
 
@@ -95,6 +96,7 @@ const (
 type ExpiringDecision struct {
 	Decision        Decision
 	Expires         time.Time
+	IpAddress       string
 	fromBaskerville bool
 }
 
@@ -133,6 +135,8 @@ type DecisionLists struct {
 	PerSiteDecisionLists StringToStringToDecision // site -> ip -> Decision
 	// dynamic lists populated from the regex rate limits + kafka
 	ExpiringDecisionLists StringToExpiringDecision // ip -> ExpiringDecision
+	// dynamic lists populated from the kafka, like ExpiringDecisionLists but session ID as index
+	ExpiringDecisionListsSessionId StringToExpiringDecision
 	// static site-wide lists (legacy banjax_sha_inv and user_banjax_sha_inv)
 	// XXX someday need sha-inv *and* captcha
 	// XXX could be merged with PerSiteDecisionLists if we matched on ip ranges
@@ -220,6 +224,7 @@ func ConfigToDecisionLists(config *Config) DecisionLists {
 	perSiteDecisionLists := make(StringToStringToDecision)
 	globalDecisionLists := make(StringToDecision)
 	expiringDecisionLists := make(StringToExpiringDecision)
+	expiringDecisionListsSessionId := make(StringToExpiringDecision)
 	sitewideShaInvList := make(StringToFailAction)
 	globalDecisionListsIPFilter := make(DecisionToIPFilter)
 	perSiteDecisionListsIPFilter := make(StringToDecisionToIPFilter)
@@ -292,7 +297,7 @@ func ConfigToDecisionLists(config *Config) DecisionLists {
 	// log.Printf("global decisions: %v\n", globalDecisionLists)
 	return DecisionLists{
 		globalDecisionLists, perSiteDecisionLists,
-		expiringDecisionLists, sitewideShaInvList,
+		expiringDecisionLists, expiringDecisionListsSessionId, sitewideShaInvList,
 		globalDecisionListsIPFilter, perSiteDecisionListsIPFilter}
 }
 
@@ -385,20 +390,6 @@ func (failedChallengeStates FailedChallengeStates) String() string {
 	return buf.String()
 }
 
-func checkExpiringDecisionLists(clientIp string, decisionLists *DecisionLists) (ExpiringDecision, bool) {
-	expiringDecision, ok := (*decisionLists).ExpiringDecisionLists[clientIp]
-	if !ok {
-		// log.Println("no mention in expiring lists")
-	} else {
-		if time.Now().Sub(expiringDecision.Expires) > 0 {
-			delete((*decisionLists).ExpiringDecisionLists, clientIp)
-			// log.Println("deleted expired decision from expiring lists")
-			ok = false
-		}
-	}
-	return expiringDecision, ok
-}
-
 // XXX mmm could hold the lock for a while?
 func RemoveExpiredDecisions(
 	decisionListsMutex *sync.Mutex,
@@ -430,19 +421,48 @@ func updateExpiringDecisionLists(
 	existingExpiringDecision, ok := (*decisionLists).ExpiringDecisionLists[ip]
 	if ok {
 		if newDecision <= existingExpiringDecision.Decision {
-			log.Println("not updating expiringDecision with less serious one", existingExpiringDecision.Decision, newDecision)
+			// log.Println("not updating expiringDecision with less serious one", existingExpiringDecision.Decision, newDecision)
 			return
 		}
 	}
 	if config.Debug {
 		log.Println("Update expiringDecision with existing and new: ", existingExpiringDecision.Decision, newDecision)
-		log.Println("From baskerville", fromBaskerville)
 	}
 
 	// XXX We are not using nginx to banjax cache feature yet
 	// purgeNginxAuthCacheForIp(ip)
 	expires := now.Add(time.Duration(config.ExpiringDecisionTtlSeconds) * time.Second)
-	(*decisionLists).ExpiringDecisionLists[ip] = ExpiringDecision{newDecision, expires, fromBaskerville}
+	(*decisionLists).ExpiringDecisionLists[ip] = ExpiringDecision{
+		newDecision, expires, ip, fromBaskerville}
+}
+
+func updateExpiringDecisionListsSessionId(
+	config *Config,
+	ip string,
+	sessionId string,
+	decisionListsMutex *sync.Mutex,
+	decisionLists *DecisionLists,
+	now time.Time,
+	newDecision Decision,
+	fromBaskerville bool,
+) {
+	decisionListsMutex.Lock()
+	defer decisionListsMutex.Unlock()
+
+	existingExpiringDecision, ok := (*decisionLists).ExpiringDecisionListsSessionId[sessionId]
+	if ok {
+		if newDecision <= existingExpiringDecision.Decision {
+			return
+		}
+	}
+
+	if config.Debug {
+		log.Printf("Update session id decision with IP %s, session id %s, existing and new: %v, %v\n",
+			ip, sessionId, existingExpiringDecision.Decision, newDecision)
+	}
+	expires := now.Add(time.Duration(config.ExpiringDecisionTtlSeconds) * time.Second)
+	(*decisionLists).ExpiringDecisionListsSessionId[sessionId] = ExpiringDecision{
+		newDecision, expires, ip, fromBaskerville}
 }
 
 type MetricsLogLine struct {

internal/http_server.go

@@ -945,7 +945,7 @@ func decisionForNginx2(
 	// changing the decision.
 	// XXX i forget if that comment is stale^
 	decisionListsMutex.Lock()
-	expiringDecision, ok := checkExpiringDecisionLists(clientIp, decisionLists)
+	expiringDecision, ok := checkExpiringDecisionLists(c, clientIp, decisionLists)
 	decisionListsMutex.Unlock()
 	if !ok {
 		// log.Println("no mention in expiring lists")
@@ -960,7 +960,7 @@ func decisionForNginx2(
 			// Check if expiringDecision.fromBaskerville, if true, check if domain disabled baskerville
 			_, disabled := config.SitesToDisableBaskerville[requestedHost]
 			if expiringDecision.fromBaskerville && disabled {
-				log.Printf("domain %s disabled baskerville, skip expiring challenge for %s", requestedHost, clientIp)
+				log.Printf("DIS-BASK: domain %s disabled baskerville, skip expiring challenge for %s", requestedHost, clientIp)
 			} else {
 				// log.Println("challenge from expiring lists")
 				sendOrValidateShaChallengeResult := sendOrValidateShaChallenge(
@@ -1032,3 +1032,30 @@ func CleanRequestedPath(requestedPath string) string {
 	path = strings.Split(path, "?")[0]
 	return path
 }
+
+func checkExpiringDecisionLists(c *gin.Context, clientIp string, decisionLists *DecisionLists) (ExpiringDecision, bool) {
+	// check session ID then check expiring lists IP
+	sessionId, err := c.Cookie(SessionCookieName)
+	if err == nil {
+		expiringDecision, ok := (*decisionLists).ExpiringDecisionListsSessionId[sessionId]
+		if ok {
+			log.Printf("DSC: found expiringDecision from session %s (%s)", sessionId, expiringDecision.Decision)
+			if time.Now().Sub(expiringDecision.Expires) > 0 {
+				delete((*decisionLists).ExpiringDecisionListsSessionId, sessionId)
+				// log.Println("deleted expired decision from expiring lists")
+				ok = false
+			}
+			return expiringDecision, ok
+		}
+	}
+
+	expiringDecision, ok := (*decisionLists).ExpiringDecisionLists[clientIp]
+	if ok {
+		if time.Now().Sub(expiringDecision.Expires) > 0 {
+			delete((*decisionLists).ExpiringDecisionLists, clientIp)
+			// log.Println("deleted expired decision from expiring lists")
+			ok = false
+		}
+	}
+	return expiringDecision, ok
+}

internal/kafka.go

@@ -12,6 +12,7 @@ import (
 	"crypto/x509"
 	"encoding/json"
 	"log"
+	"net/url"
 	"os"
 	"sync"
 	"time"
@@ -157,12 +158,51 @@ func handleCommand(
 				Challenge,
 				true, // from baskerville, provide to http_server to distinguish from regex
 			)
-			log.Printf("KAFKA: added to global challenge lists: Challenge %s\n", command.Value)
+			log.Printf("KAFKA: challenge_ip: %s\n", command.Value)
 		} else if disabled {
-			log.Printf("KAFKA: not challenge %s, site %s disables baskerville\n", command.Value, command.Host)
+			log.Printf("KAFKA: DIS-BASK: not challenge %s, site %s disabled baskerville\n", command.Value, command.Host)
 		} else {
 			log.Printf("KAFKA: command value looks malformed: %s\n", command.Value)
 		}
+		break
+	case "challenge_session":
+	case "block_session":
+		if command.SessionId == "" {
+			log.Printf("KAFKA: session_id is EMPTY, break\n")
+			break
+		}
+		// exempt a site from challenge according to config
+		_, disabled := config.SitesToDisableBaskerville[command.Host]
+
+		if !disabled {
+			// gin does urldecode or cookie, so we decode any possible urlencoded session id from kafka
+			sessionIdDecoded, decodeErr := url.QueryUnescape(command.SessionId)
+			if decodeErr != nil {
+				log.Printf("KAFKA: fail to urldecode session_id %s, break\n", command.SessionId)
+				break
+			}
+			var decision Decision
+			if command.Name == "block_session" {
+				log.Printf("KAFKA: block_session: %s\n", sessionIdDecoded)
+				decision = NginxBlock
+			} else {
+				log.Printf("KAFKA: challenge_session: %s\n", sessionIdDecoded)
+				decision = Challenge
+			}
+			updateExpiringDecisionListsSessionId(
+				config,
+				command.Value,
+				sessionIdDecoded,
+				decisionListsMutex,
+				decisionLists,
+				time.Now(),
+				decision,
+				true, // from baskerville, provide to http_server to distinguish from regex
+			)
+		} else {
+			log.Printf("KAFKA: DIS-BASK: no action on %s, site %s disabled baskerville\n", command.Value, command.Host)
+		}
+		break
 	default:
 		log.Printf("KAFKA: unrecognized command name: %s\n", command.Name)
 	}

internal/session_cookie.go

@@ -131,7 +131,7 @@ func sessionCookieEndPoint(c *gin.Context, config *Config) error {
 	if err == nil {
 		// cookie exists, validate it
 		validateErr := validateSessionCookie(urlDecodedDsc, config.SessionCookieHmacSecret, time.Now(), clientIp)
-		if validateErr == nil {
+		if validateErr == nil || config.SessionCookieNotVerify {
 			// cookie is valid, do not attach cookie but only report dsc_new=false
 			// log.Printf("DSC: [%s] cookie %s is valid, report dsc_new=false\n", clientIp, urlDecodedDsc)
 			attachSessionCookie(c, config, urlDecodedDsc, false)