first version

Makefile

@@ -0,0 +1,59 @@
+NAME     = kf2-antiddos
+VERSION := $(shell git describe)
+GOCMD    = go
+LDFLAGS := "$(LDFLAGS) -s -w -X 'main.Version=$(VERSION)'"
+GOBUILD  = $(GOCMD) build -ldflags=$(LDFLAGS)
+SRCMAIN  = ./cmd/$(NAME)
+SRCDOC   = ./doc
+BINDIR   = bin
+BIN      = $(BINDIR)/$(NAME)
+README   = $(SRCDOC)/README
+LICENSE  = LICENSE
+PREFIX   = /usr
+
+.PHONY: all prep doc build check-build linux-amd64 windows-amd64 compile install check-install uninstall clean
+
+all: build
+
+prep: clean
+	go mod init $(NAME); go mod tidy
+	mkdir $(BINDIR)
+	
+doc: check-build
+	test -d $(SRCDOC) || mkdir $(SRCDOC)
+	$(BIN) --help > $(README)
+	
+build: prep
+	$(GOBUILD) -o $(BIN) $(SRCMAIN)
+
+check-build:
+	test -e $(BIN)
+
+linux-amd64: prep
+	GOOS=linux GOARCH=amd64 $(GOBUILD) -o $(BIN)-linux-amd64 $(SRCMAIN)
+
+windows-amd64: prep
+	GOOS=windows GOARCH=amd64 $(GOBUILD) -o $(BIN)-windows-amd64.exe $(SRCMAIN)
+
+compile: linux-386 windows-386 linux-amd64 windows-amd64
+	
+install: check-build doc
+	install -m 755 -d         $(PREFIX)/bin/
+	install -m 755 $(BIN)     $(PREFIX)/bin/
+	install -m 755 -d         $(PREFIX)/share/licenses/$(NAME)/
+	install -m 644 $(LICENSE) $(PREFIX)/share/licenses/$(NAME)/
+	install -m 755 -d         $(PREFIX)/share/doc/$(NAME)/
+	install -m 644 $(README)  $(PREFIX)/share/doc/$(NAME)/
+
+check-install:
+	test -e $(PREFIX)/bin/$(NAME) || \
+	test -d $(PREFIX)/share/licenses/$(NAME) || \
+	test -d $(PREFIX)/share/doc/$(NAME)
+
+uninstall: check-install
+	rm -f  $(PREFIX)/bin/$(NAME)
+	rm -rf $(PREFIX)/share/licenses/$(NAME)
+	rm -rf $(PREFIX)/share/doc/$(NAME)
+
+clean:
+	rm -rf $(BINDIR)

cmd/kf2-antiddos/args.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"github.com/juju/gnuflag"
+
+	"kf2-antiddos/internal/config"
+	"kf2-antiddos/internal/output"
+)
+
+func printHelp() {
+	output.Println("Anti DDoS tool for kf2 servers")
+	output.Println("")
+	output.Printf("Usage: <kf2_logs_output> | %s [option]... <shell> <deny_script> <allow_script>", AppName)
+	output.Println("kf2_logs_output            KF2 logs to redirect to stdin")
+	output.Println("shell                      shell to run deny_script and allow_script")
+	output.Println("deny_script                firewall deny script (takes IP as argument)")
+	output.Println("allow_script               firewall allow script (takes IPs as arguments)")
+	output.Println("")
+	output.Println("Options:")
+	output.Println("  -j, --jobs N             allow N jobs at once")
+	output.Println("  -o, --output MODE        self|proxy|all|quiet")
+	output.Println(" -dt, --deny-time TIME     minimum ip deny TIME (seconds)")
+	output.Println(" -mc, --max-connections N  Skip N connections before run deny script")
+	output.Println("  -v, --version            Show version")
+	output.Println("  -h, --help               Show help")
+}
+
+func printVersion() {
+	output.Printf("%s %s", AppName, AppVersion)
+}
+
+func parseArgs() config.Config {
+	rawCfg := config.Config{}
+
+	gnuflag.UintVar(&rawCfg.Jobs, "j", 0, "")
+	gnuflag.UintVar(&rawCfg.Jobs, "jobs", 0, "")
+
+	gnuflag.StringVar(&rawCfg.OutputMode, "o", "", "")
+	gnuflag.StringVar(&rawCfg.OutputMode, "output", "", "")
+
+	gnuflag.UintVar(&rawCfg.DenyTime, "dt", 0, "")
+	gnuflag.UintVar(&rawCfg.DenyTime, "deny-time", 0, "")
+
+	gnuflag.UintVar(&rawCfg.MaxConn, "mc", 0, "")
+	gnuflag.UintVar(&rawCfg.MaxConn, "max-connections", 0, "")
+
+	gnuflag.BoolVar(&rawCfg.ShowVersion, "v", false, "")
+	gnuflag.BoolVar(&rawCfg.ShowVersion, "version", false, "")
+
+	gnuflag.BoolVar(&rawCfg.ShowHelp, "h", false, "")
+	gnuflag.BoolVar(&rawCfg.ShowHelp, "help", false, "")
+
+	gnuflag.Parse(true)
+
+	for i := 0; i < 3 && i < gnuflag.NArg(); i++ {
+		switch i {
+		case 0:
+			rawCfg.Shell = gnuflag.Arg(i)
+		case 1:
+			rawCfg.DenyAction = gnuflag.Arg(i)
+		case 2:
+			rawCfg.AllowAction = gnuflag.Arg(i)
+		}
+	}
+
+	return rawCfg
+}

cmd/kf2-antiddos/main.go

@@ -0,0 +1,140 @@
+package main
+
+import (
+	"kf2-antiddos/internal/action"
+	"kf2-antiddos/internal/common"
+	"kf2-antiddos/internal/config"
+	"kf2-antiddos/internal/history"
+	"kf2-antiddos/internal/output"
+	"kf2-antiddos/internal/parser"
+	"kf2-antiddos/internal/reader"
+
+	"os"
+	"os/signal"
+	"runtime"
+	"sync"
+	"syscall"
+)
+
+const (
+	ExitSuccess  int = 0
+	ExitArgError int = 1
+)
+
+const (
+	AppName = "kf2-antiddos"
+)
+
+var (
+	AppVersion string = "dev"
+)
+
+func main() {
+	cfg := parseArgs()
+
+	switch {
+	case cfg.ShowHelp:
+		printHelp()
+		os.Exit(ExitSuccess)
+	case cfg.ShowVersion:
+		printVersion()
+		os.Exit(ExitSuccess)
+	}
+
+	if cfg.IsValid() {
+		cfg.SetEmptyArgs()
+	} else {
+		os.Exit(ExitArgError)
+	}
+
+	switch cfg.OutputMode {
+	case config.OT_All:
+		output.AllMode()
+	case config.OT_Proxy:
+		output.ProxyMode()
+	case config.OT_Quiet:
+		output.QuietMode()
+	case config.OT_Self:
+		output.SelfMode()
+	}
+
+	runtime.GOMAXPROCS(int(cfg.Jobs))
+
+	Workers := make([]common.Worker, 0, cfg.Jobs+3)
+
+	wg := sync.WaitGroup{}
+
+	// Data flow:
+	banChan := make(chan string, cfg.Jobs)
+	inputChan := make(chan common.RawEvent, cfg.Jobs)
+	eventChan := make(chan common.Event, cfg.Jobs)
+	resetChan := make(chan string, cfg.Jobs)
+
+	// Reader worker
+	Workers = append(Workers,
+		reader.New(
+			uint(len(Workers)),
+			&inputChan,
+		))
+
+	// parser workers
+	for i := uint(0); i < cfg.Jobs; i++ {
+		Workers = append(Workers,
+			parser.New(
+				uint(len(Workers)),
+				&inputChan,
+				&eventChan,
+			))
+	}
+
+	// History worker
+	Workers = append(Workers,
+		history.New(
+			uint(len(Workers)),
+			&eventChan,
+			&banChan,
+			&resetChan,
+			cfg.MaxConn,
+		))
+
+	// Action worker
+	Workers = append(Workers,
+		action.New(
+			uint(len(Workers)),
+			cfg.DenyTime,
+			cfg.Shell,
+			cfg.AllowAction,
+			cfg.DenyAction,
+			&banChan,
+			&resetChan,
+		))
+
+	wg.Add(len(Workers))
+
+	closeHandler(Workers, &wg)
+
+	for i := range Workers {
+		Workers[i].Do()
+	}
+
+	output.Println("started")
+
+	wg.Wait()
+
+	output.Println("exit")
+
+	os.Exit(ExitSuccess)
+}
+
+func closeHandler(Workers []common.Worker, wg *sync.WaitGroup) {
+	interrupt := make(chan os.Signal, 1)
+	signal.Notify(interrupt, syscall.SIGINT, syscall.SIGTERM, os.Interrupt)
+	go func() {
+		<-interrupt
+		output.Println("interrupt")
+		for _, worker := range Workers {
+			worker.Stop()
+			wg.Done()
+		}
+	}()
+}

doc/README

@@ -0,0 +1,15 @@
+Anti DDoS tool for kf2 servers
+
+Usage: <kf2_logs_output> | kf2-antiddos [option]... <shell> <deny_script> <allow_script>
+kf2_logs_output            KF2 logs to redirect to stdin
+shell                      shell to run deny_script and allow_script
+deny_script                firewall deny script (takes IP as argument)
+allow_script               firewall allow script (takes IPs as arguments)
+
+Options:
+  -j, --jobs N             allow N jobs at once
+  -o, --output MODE        self|proxy|all|quiet
+ -dt, --deny-time TIME     minimum ip deny TIME (seconds)
+ -mc, --max-connections N  Skip N connections before run deny script
+  -v, --version            Show version
+  -h, --help               Show help

go.mod

@@ -0,0 +1,5 @@
+module kf2-antiddos
+
+go 1.15
+
+require github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d

go.sum

@@ -0,0 +1,2 @@
+github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d h1:c93kUJDtVAXFEhsCh5jSxyOJmFHuzcihnslQiX8Urwo=
+github.com/juju/gnuflag v0.0.0-20171113085948-2ce1bb71843d/go.mod h1:2PavIy+JPciBPrBUjwbNvtwB6RQlve+hkpll6QSNmOE=

internal/action/action.go

@@ -0,0 +1,118 @@
+package action
+
+import (
+	"kf2-antiddos/internal/output"
+
+	"os"
+	"os/exec"
+	"strings"
+	"time"
+)
+
+type Action struct {
+	ticker      *time.Ticker
+	ips         map[string]bool // map[IP]readyToUnban
+	allowAction string
+	denyAction  string
+	shell       string
+	quit        chan struct{}
+	banChan     *chan string
+	resetChan   *chan string
+	workerID    uint
+}
+
+func New(workerID uint, denyTime uint, shell, allowAction, denyAction string, banChan, resetChan *chan string) *Action {
+	return &Action{
+		ticker:      time.NewTicker(time.Duration(denyTime) * time.Second),
+		ips:         make(map[string]bool),
+		allowAction: allowAction,
+		denyAction:  denyAction,
+		shell:       shell,
+		quit:        make(chan struct{}),
+		banChan:     banChan,
+		resetChan:   resetChan,
+		workerID:    workerID,
+	}
+}
+
+func (a *Action) Do() {
+	go func() {
+		for {
+			select {
+			case ip := <-*a.banChan:
+				a.deny(ip)
+			case <-a.ticker.C:
+				a.allow(false)
+			case <-a.quit:
+				a.ticker.Stop()
+				a.allow(true)
+				return
+			}
+		}
+	}()
+}
+
+func (a *Action) Stop() {
+	close(a.quit)
+}
+
+func (a *Action) allow(unbanAll bool) {
+	unban := make([]string, 0)
+
+	for ip := range a.ips {
+		if unbanAll || bool(a.ips[ip]) { // aka if readyToUnban
+			unban = append(unban, ip)
+		} else {
+			a.ips[ip] = true // mark readyToUnban next time
+		}
+	}
+
+	for _, ip := range unban {
+		delete(a.ips, ip)
+	}
+
+	if len(unban) != 0 {
+		for _, ip := range unban {
+			*a.resetChan <- ip
+		}
+		output.Printf("Allow: %s", strings.Join(unban, ", "))
+
+		if err := a.execCmd(a.allowAction, unban); err != nil {
+			output.Error(err.Error())
+			return
+		}
+	}
+}
+
+func (a *Action) deny(ip string) {
+	a.ips[ip] = false
+
+	output.Printf("Ban: %s", ip)
+
+	if err := a.execCmd(a.denyAction, []string{ip}); err != nil {
+		output.Error(err.Error())
+		return
+	}
+}
+
+func (a *Action) execCmd(command string, args []string) error {
+	WorkingDir, err := os.Getwd()
+	if err != nil {
+		WorkingDir = ""
+	}
+	cmd := &exec.Cmd{
+		Path:   a.shell,
+		Args:   append([]string{a.shell, command}, args...),
+		Stdout: output.StdoutWriter(),
+		Stderr: output.StderrWriter(),
+		Dir:    WorkingDir,
+	}
+
+	if err := cmd.Start(); err != nil {
+		return err
+	}
+	if err := cmd.Wait(); err != nil {
+		return err
+	}
+	return nil
+}

internal/common/common.go

@@ -0,0 +1,18 @@
+package common
+
+type Worker interface {
+	Do()
+	Stop()
+}
+
+type RawEvent struct {
+	LineNum byte
+	Text    string
+}
+
+type Event struct {
+	LineNum       byte
+	ConnectIP     string
+	PlayerStartIP string
+	PlayerEndIP   string
+}

internal/config/config.go

@@ -0,0 +1,82 @@
+package config
+
+import (
+	"fmt"
+	"os"
+	"runtime"
+
+	"kf2-antiddos/internal/output"
+)
+
+const (
+	OT_Proxy = "proxy"
+	OT_Self  = "self"
+	OT_All   = "all"
+	OT_Quiet = "quiet"
+)
+
+type Config struct {
+	Shell       string
+	DenyAction  string
+	AllowAction string
+	Jobs        uint
+	OutputMode  string
+	DenyTime    uint
+	MaxConn     uint
+
+	ShowVersion bool
+	ShowHelp    bool
+}
+
+func (cfg Config) IsValid() bool {
+	errs := make([]string, 0)
+
+	if cfg.Shell == "" {
+		errs = append(errs, "shell can not be empty")
+	} else if _, err := os.Stat(cfg.Shell); os.IsNotExist(err) {
+		errs = append(errs, fmt.Sprintf("shell %s not found", cfg.Shell))
+	}
+
+	if cfg.AllowAction == "" {
+		errs = append(errs, "allow_action can not be empty")
+	} else if _, err := os.Stat(cfg.AllowAction); os.IsNotExist(err) {
+		errs = append(errs, fmt.Sprintf("allow_action file %s not found", cfg.AllowAction))
+	}
+
+	if cfg.DenyAction == "" {
+		errs = append(errs, "deny_action can not be empty")
+	} else if _, err := os.Stat(cfg.DenyAction); os.IsNotExist(err) {
+		errs = append(errs, fmt.Sprintf("deny_action file %s not found", cfg.DenyAction))
+	}
+
+	switch cfg.OutputMode {
+	case OT_Proxy:
+	case OT_Self:
+	case OT_All:
+	case OT_Quiet:
+	case "":
+	default:
+		errs = append(errs, fmt.Sprintf("Unknown output_type: %s", cfg.OutputMode))
+	}
+
+	for _, err := range errs {
+		output.Errorln(err)
+	}
+
+	return len(errs) == 0
+}
+
+func (cfg *Config) SetEmptyArgs() {
+	if cfg.Jobs == 0 {
+		cfg.Jobs = uint(runtime.NumCPU())
+	}
+	if cfg.MaxConn == 0 {
+		cfg.MaxConn = 10
+	}
+	if cfg.OutputMode == "" {
+		cfg.OutputMode = OT_Self
+	}
+	if cfg.DenyTime == 0 {
+		cfg.DenyTime = 20 * 60
+	}
+}

internal/history/history.go

@@ -0,0 +1,103 @@
+package history
+
+import (
+	"kf2-antiddos/internal/common"
+)
+
+type History struct {
+	quit      chan struct{}
+	eventChan *chan common.Event
+	banChan   *chan string
+	resetChan *chan string
+	head      byte
+	history   map[byte]common.Event
+	ips       map[string]uint // map[ip]conn_count
+	whitelist map[string]struct{}
+	banned    map[string]struct{}
+	maxConn   uint
+	workerID  uint
+}
+
+func New(workerID uint, eventChan *chan common.Event, banChan *chan string, resetChan *chan string, maxConn uint) *History {
+	return &History{
+		quit:      make(chan struct{}),
+		ips:       make(map[string]uint, 0),
+		history:   make(map[byte]common.Event, 0),
+		whitelist: make(map[string]struct{}, 0),
+		banned:    make(map[string]struct{}, 0),
+		eventChan: eventChan,
+		banChan:   banChan,
+		resetChan: resetChan,
+		head:      0,
+		maxConn:   maxConn,
+		workerID:  workerID,
+	}
+}
+
+func (h *History) Do() {
+	go func() {
+		for {
+			select {
+			case event := <-*h.eventChan:
+				h.registerEvent(event)
+			case ip := <-*h.resetChan:
+				h.resetIp(ip)
+			case <-h.quit:
+				return
+			}
+		}
+	}()
+}
+
+func (h *History) Stop() {
+	close(h.quit)
+}
+
+func (h *History) registerEvent(e common.Event) {
+	h.history[e.LineNum] = e
+
+	for {
+		nextEvent, nextEventExists := h.history[h.head+1]
+		if nextEventExists {
+			switch {
+			case nextEvent.ConnectIP != "":
+				h.registerConnect(nextEvent.ConnectIP)
+			case nextEvent.PlayerEndIP != "":
+				h.registerEndPlayer(nextEvent.PlayerEndIP)
+			case nextEvent.PlayerStartIP != "":
+				h.registerNewPlayer(nextEvent.PlayerEndIP)
+			}
+			delete(h.history, h.head+1)
+			h.head++
+		} else {
+			break
+		}
+	}
+}
+
+func (h *History) registerConnect(ip string) {
+	h.ips[ip]++
+	if h.ips[ip] > h.maxConn {
+		_, whitelisted := h.whitelist[ip]
+		_, banned := h.banned[ip]
+		if !whitelisted && !banned {
+			h.banned[ip] = struct{}{}
+			*h.banChan <- ip
+		}
+	}
+}
+
+func (h *History) registerNewPlayer(ip string) {
+	h.whitelist[ip] = struct{}{}
+}
+
+func (h *History) registerEndPlayer(ip string) {
+	delete(h.whitelist, ip)
+	delete(h.ips, ip)
+	delete(h.banned, ip)
+}
+
+func (h *History) resetIp(ip string) {
+	delete(h.ips, ip)
+	delete(h.banned, ip)
+}

internal/output/output.go

@@ -0,0 +1,111 @@
+package output
+
+import (
+	"fmt"
+	"io"
+	"io/ioutil"
+	"log"
+	"os"
+	"runtime"
+)
+
+const (
+	AppName = "[kf2-antiddos] "
+)
+
+var (
+	endOfLine string      = "\n"
+	devNull   *log.Logger = log.New(ioutil.Discard, "", 0)
+	stdout    *log.Logger = log.New(os.Stdout, "", 0)
+	stderr    *log.Logger = log.New(os.Stderr, "", 0)
+	proxy     *log.Logger = log.New(os.Stdout, "", 0)
+)
+
+func ProxyMode() {
+	stdout = devNull
+	stderr = devNull
+	proxy = log.New(os.Stdout, "", 0)
+}
+
+func SelfMode() {
+	proxy = devNull
+	stdout = log.New(os.Stdout, "", 0)
+	stderr = log.New(os.Stderr, "", 0)
+}
+
+func AllMode() {
+	stdout = log.New(os.Stdout, AppName, 0)
+	stderr = log.New(os.Stderr, AppName, 0)
+	proxy = log.New(os.Stdout, "", 0)
+}
+
+func StdoutWriter() io.Writer {
+	return stdout.Writer()
+}
+
+func StderrWriter() io.Writer {
+	return stderr.Writer()
+}
+
+func QuietMode() {
+	stdout = devNull
+	stderr = devNull
+	proxy = devNull
+}
+
+func SetEndOfLineNative() {
+	switch os := runtime.GOOS; os {
+	case "windows":
+		setEndOfLineWindows()
+	default:
+		setEndOfLineUnix()
+	}
+}
+
+func EOL() string {
+	return endOfLine
+}
+
+func setEndOfLineUnix() {
+	endOfLine = "\n"
+}
+
+func setEndOfLineWindows() {
+	endOfLine = "\r\n"
+}
+
+func Print(v ...interface{}) {
+	stdout.Print(v...)
+}
+
+func Printf(format string, v ...interface{}) {
+	stdout.Printf(format, v...)
+}
+
+func Println(v ...interface{}) {
+	stdout.Print(fmt.Sprint(v...) + endOfLine)
+}
+
+func Error(v ...interface{}) {
+	stderr.Print(v...)
+}
+
+func Errorf(format string, v ...interface{}) {
+	stderr.Printf(format, v...)
+}
+
+func Errorln(v ...interface{}) {
+	stderr.Print(fmt.Sprint(v...) + endOfLine)
+}
+
+func Proxy(v ...interface{}) {
+	proxy.Print(v...)
+}
+
+func Proxyf(format string, v ...interface{}) {
+	proxy.Printf(format, v...)
+}
+
+func Proxyln(v ...interface{}) {
+	proxy.Print(fmt.Sprint(v...) + endOfLine)
+}

internal/parser/parser.go

@@ -0,0 +1,75 @@
+package parser
+
+import (
+	"kf2-antiddos/internal/common"
+	"regexp"
+)
+
+const (
+	ngConnectIP     = "ConnectIP"
+	ngPlayerStartIP = "PlayerStartIP"
+	ngPlayerEndIP   = "PlayerEndIP"
+	rxIP            = `\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}`
+	rxConnect       = `NetComeGo:\sOpen\sTheWorld\s+(?P<` + ngConnectIP + `>` + rxIP + `){1}`
+	rxPlayerStart   = `DevOnline:\sVerifyClientAuthSession:\sClientIP:\s(?P<` + ngPlayerStartIP + `>` + rxIP + `){1}`
+	rxPlayerEnd     = `DevOnline:\sEndRemoteClientAuthSession:\sClientAddr:\s(?P<` + ngPlayerEndIP + `>` + rxIP + `){1}`
+	rxValue         = rxConnect + `|` + rxPlayerStart + `|` + rxPlayerEnd
+)
+
+var (
+	rxKFLog *regexp.Regexp = regexp.MustCompile(rxValue)
+)
+
+type Parser struct {
+	quit       chan struct{}
+	inputChan  *chan common.RawEvent
+	outputChan *chan common.Event
+	workerID   uint
+}
+
+func New(workerID uint, inputChan *chan common.RawEvent, outputChan *chan common.Event) *Parser {
+	return &Parser{
+		inputChan:  inputChan,
+		outputChan: outputChan,
+		quit:       make(chan struct{}),
+		workerID:   workerID,
+	}
+}
+
+func (p *Parser) Do() {
+	go func() {
+		for {
+			select {
+			case rawEvent := <-*p.inputChan:
+				*p.outputChan <- p.parse(rawEvent)
+			case <-p.quit:
+				return
+			}
+		}
+	}()
+}
+
+func (p *Parser) Stop() {
+	close(p.quit)
+}
+
+func (p *Parser) parse(rawEvent common.RawEvent) common.Event {
+	res := common.Event{
+		LineNum: rawEvent.LineNum,
+	}
+
+	match := rxKFLog.FindStringSubmatch(rawEvent.Text)
+	for i, name := range rxKFLog.SubexpNames() {
+		if i != 0 && name != "" && i <= len(match) && match[i] != "" {
+			switch name {
+			case ngConnectIP:
+				res.ConnectIP = match[i]
+			case ngPlayerStartIP:
+				res.PlayerStartIP = match[i]
+			case ngPlayerEndIP:
+				res.PlayerEndIP = match[i]
+			}
+		}
+	}
+	return res
+}

internal/reader/reader.go

@@ -0,0 +1,59 @@
+package reader
+
+import (
+	"kf2-antiddos/internal/common"
+	"kf2-antiddos/internal/output"
+
+	"bufio"
+	"os"
+)
+
+type Reader struct {
+	quit       chan struct{}
+	outputChan *chan common.RawEvent
+	workerID   uint
+}
+
+func New(workerID uint, outputChan *chan common.RawEvent) *Reader {
+	return &Reader{
+		outputChan: outputChan,
+		quit:       make(chan struct{}),
+		workerID:   workerID,
+	}
+}
+
+func (r *Reader) Do() {
+	go func() {
+		stdin := bufio.NewScanner(os.Stdin)
+		stdin.Split(bufio.ScanLines)
+		for {
+			select {
+			case <-r.quit: // check quit if there are no input
+				return
+			default:
+				for lineNum := byte(1); stdin.Scan(); lineNum++ { // byte overflow it's not a bug, but a feature
+					select {
+					case <-r.quit: // check quit if there are input
+						return
+					default:
+					}
+
+					text := stdin.Text()
+					output.Proxyln(text)
+					*r.outputChan <- common.RawEvent{
+						LineNum: lineNum,
+						Text:    text,
+					}
+				}
+
+				if err := stdin.Err(); err != nil {
+					output.Errorln(err.Error())
+				}
+			}
+		}
+	}()
+}
+
+func (r *Reader) Stop() {
+	close(r.quit)
+}