rename module to hook

8 files changed, 664 insertions, 0 deletions

diff --git a/pkg/server/hook/aes.go b/pkg/server/hook/aes.go
new file mode 100644
index 0000000..b461a34
--- /dev/null
+++ b/pkg/server/hook/aes.go
@@ -0,0 +1,87 @@
+package hook
+
+import (
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/md5"
+	"crypto/rand"
+	"io"
+	"tunnel/pkg/server/env"
+	"tunnel/pkg/server/opts"
+	"tunnel/pkg/server/queue"
+)
+
+type aesHook struct{}
+
+type aesPipe struct {
+	key []byte
+}
+
+func (a *aesPipe) Send(rq, wq queue.Q) error {
+	block, err := aes.NewCipher(a.key)
+	if err != nil {
+		return err
+	}
+
+	iv := make([]byte, aes.BlockSize)
+
+	if _, err := rand.Read(iv); err != nil {
+		return err
+	}
+
+	writer := &cipher.StreamWriter{
+		S: cipher.NewOFB(block, iv),
+		W: wq.Writer(),
+	}
+
+	wq <- iv
+
+	return queue.IoCopy(rq.Reader(), writer)
+}
+
+func (a *aesPipe) Recv(rq, wq queue.Q) error {
+	block, err := aes.NewCipher(a.key)
+	if err != nil {
+		return err
+	}
+
+	r := rq.Reader()
+
+	iv := make([]byte, aes.BlockSize)
+
+	if _, err := io.ReadFull(r, iv); err != nil {
+		if err == io.EOF {
+			return nil
+		}
+		return err
+	}
+
+	reader := &cipher.StreamReader{
+		S: cipher.NewOFB(block, iv),
+		R: r,
+	}
+
+	return queue.IoCopy(reader, wq.Writer())
+}
+
+func newAes(env env.Env) *aesPipe {
+	s := getAuthSecret(env)
+	h := md5.Sum([]byte(s))
+
+	a := &aesPipe{key: make([]byte, 16)}
+	copy(a.key, h[:])
+
+	return a
+}
+
+func (h aesHook) Open(env env.Env) (interface{}, error) {
+	return newAes(env), nil
+}
+
+func newAesHook(opts.Opts, env.Env) (hook, error) {
+	return aesHook{}, nil
+}
+
+func init() {
+	register("aes", newAesHook)
+}
diff --git a/pkg/server/hook/alpha.go b/pkg/server/hook/alpha.go
new file mode 100644
index 0000000..d1fefcc
--- /dev/null
+++ b/pkg/server/hook/alpha.go
@@ -0,0 +1,32 @@
+package hook
+
+import (
+	"bufio"
+	"io"
+	"tunnel/pkg/server/queue"
+	"unicode"
+)
+
+func alpha(cb func(rune) rune) Func {
+	return func(rq, wq queue.Q) error {
+		r := bufio.NewReader(rq.Reader())
+
+		for {
+			c, _, err := r.ReadRune()
+			if err != nil {
+				if err == io.EOF {
+					break
+				}
+				return err
+			}
+			wq <- []byte(string(cb(c)))
+		}
+
+		return nil
+	}
+}
+
+func init() {
+	registerFunc("lower", alpha(unicode.ToLower))
+	registerFunc("upper", alpha(unicode.ToUpper))
+}
diff --git a/pkg/server/hook/auth.go b/pkg/server/hook/auth.go
new file mode 100644
index 0000000..fc19c2a
--- /dev/null
+++ b/pkg/server/hook/auth.go
@@ -0,0 +1,150 @@
+package hook
+
+import (
+	"crypto/md5"
+	"crypto/rand"
+	"errors"
+	"io"
+	"tunnel/pkg/netstring"
+	"tunnel/pkg/server/env"
+	"tunnel/pkg/server/opts"
+	"tunnel/pkg/server/queue"
+)
+
+const ChallengeLen = 16
+
+type auth struct {
+	secret string
+
+	challenge struct {
+		self string
+		peer string
+	}
+
+	hash string
+
+	recvChallenge chan struct{}
+	recvHash      chan struct{}
+
+	fail chan struct{}
+	ok   chan struct{}
+}
+
+var errDupChallenge = errors.New("peer duplicates challenge")
+var errAuthFail = errors.New("peer auth fail")
+
+type authHook struct{}
+
+func (a *auth) generateChallenge() error {
+	b := make([]byte, ChallengeLen)
+	if _, err := rand.Read(b); err != nil {
+		return err
+	}
+
+	a.challenge.self = string(b)
+
+	return nil
+}
+
+func (a *auth) getHash(c string) string {
+	h := md5.New()
+
+	io.WriteString(h, a.secret)
+	io.WriteString(h, c)
+
+	return string(h.Sum(nil))
+}
+
+func (a *auth) isReady(c chan struct{}) bool {
+	select {
+	case <-a.fail:
+		return false
+	case <-c:
+		return true
+	}
+}
+
+func (a *auth) Send(rq, wq queue.Q) error {
+	e := netstring.NewEncoder(wq.Writer())
+
+	if err := a.generateChallenge(); err != nil {
+		return err
+	}
+
+	e.Encode(a.challenge.self)
+
+	if !a.isReady(a.recvChallenge) {
+		return nil
+	}
+
+	if a.challenge.self == a.challenge.peer {
+		return errDupChallenge
+	}
+
+	e.Encode(a.getHash(a.challenge.peer))
+
+	if !a.isReady(a.recvHash) {
+		return nil
+	}
+
+	if a.hash != a.getHash(a.challenge.self) {
+		close(a.fail)
+		return errAuthFail
+	}
+
+	close(a.ok)
+
+	return queue.Copy(rq, wq)
+}
+
+func (a *auth) Recv(rq, wq queue.Q) (err error) {
+	r := rq.Reader()
+	d := netstring.NewDecoder(r)
+
+	if a.challenge.peer, err = d.Decode(); err != nil {
+		close(a.fail)
+		return
+	}
+
+	close(a.recvChallenge)
+
+	if a.hash, err = d.Decode(); err != nil {
+		close(a.fail)
+		return err
+	}
+
+	close(a.recvHash)
+
+	if !a.isReady(a.ok) {
+		return nil
+	}
+
+	return queue.IoCopy(r, wq.Writer())
+}
+
+func getAuthSecret(env env.Env) string {
+	if v := env.Eval("@{tunnel.@{tunnel}.secret}"); v != "" {
+		return v
+	}
+
+	return env.Get("secret")
+}
+
+func (h authHook) Open(env env.Env) (interface{}, error) {
+	a := &auth{
+		secret:        getAuthSecret(env),
+		recvChallenge: make(chan struct{}),
+		recvHash:      make(chan struct{}),
+		fail:          make(chan struct{}),
+		ok:            make(chan struct{}),
+	}
+	return a, nil
+}
+
+func newAuthHook(opts.Opts, env.Env) (hook, error) {
+	return authHook{}, nil
+}
+
+func init() {
+	register("auth", newAuthHook)
+}
diff --git a/pkg/server/hook/hex.go b/pkg/server/hook/hex.go
new file mode 100644
index 0000000..beaadeb
--- /dev/null
+++ b/pkg/server/hook/hex.go
@@ -0,0 +1,37 @@
+package hook
+
+import (
+	"encoding/hex"
+	"tunnel/pkg/server/env"
+	"tunnel/pkg/server/opts"
+	"tunnel/pkg/server/queue"
+)
+
+type hexHook struct{}
+
+func (h hexHook) Send(rq, wq queue.Q) error {
+	enc := hex.NewEncoder(wq.Writer())
+
+	for b := range rq {
+		enc.Write(b)
+	}
+
+	return nil
+}
+
+func (h hexHook) Recv(rq, wq queue.Q) error {
+	r := hex.NewDecoder(rq.Reader())
+	return queue.IoCopy(r, wq.Writer())
+}
+
+func (h hexHook) Open(env.Env) (interface{}, error) {
+	return h, nil
+}
+
+func newHexHook(opts.Opts, env.Env) (hook, error) {
+	return hexHook{}, nil
+}
+
+func init() {
+	register("hex", newHexHook)
+}
diff --git a/pkg/server/hook/hook.go b/pkg/server/hook/hook.go
new file mode 100644
index 0000000..b2970ac
--- /dev/null
+++ b/pkg/server/hook/hook.go
@@ -0,0 +1,126 @@
+package hook
+
+import (
+	"fmt"
+	"log"
+	"sort"
+	"strings"
+	"tunnel/pkg/server/env"
+	"tunnel/pkg/server/opts"
+	"tunnel/pkg/server/queue"
+)
+
+type hookInitFunc func(opts.Opts, env.Env) (hook, error)
+
+var hooks = map[string]hookInitFunc{}
+
+type hook interface {
+	Open(env env.Env) (interface{}, error)
+}
+
+type H interface {
+	hook
+	String() string
+}
+
+type Sender interface {
+	Send(rq, wq queue.Q) error
+}
+
+type Recver interface {
+	Recv(rq, wq queue.Q) error
+}
+
+type Func func(rq, wq queue.Q) error
+
+func (f Func) Send(rq, wq queue.Q) error {
+	return f(rq, wq)
+}
+
+func (f Func) Open(env env.Env) (interface{}, error) {
+	return f, nil
+}
+
+type wrapper struct {
+	hook
+	name    string
+	reverse bool
+}
+
+func (w *wrapper) String() string {
+	return fmt.Sprintf("hook:%s", w.name)
+}
+
+func Open(h H, env env.Env) (Func, Func, error) {
+	var send, recv Func
+
+	w := h.(*wrapper)
+
+	it, err := h.Open(env)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if sender, ok := it.(Sender); ok {
+		send = sender.Send
+	}
+
+	if recver, ok := it.(Recver); ok {
+		recv = recver.Recv
+	}
+
+	if w.reverse {
+		send, recv = recv, send
+	}
+
+	return send, recv, nil
+}
+
+func New(desc string, env env.Env) (H, error) {
+	name, opts := opts.Parse(desc)
+	reverse := false
+
+	if strings.HasPrefix(name, "-") {
+		name = name[1:]
+		reverse = true
+	}
+
+	if f, ok := hooks[name]; !ok {
+		return nil, fmt.Errorf("unknown hook '%s'", name)
+	} else if h, err := f(opts, env); err != nil {
+		return nil, err
+	} else {
+		w := &wrapper{
+			hook:    h,
+			name:    name,
+			reverse: reverse,
+		}
+		return w, nil
+	}
+}
+
+func register(name string, f hookInitFunc) {
+	if _, ok := hooks[name]; ok {
+		log.Panicf("duplicate hook name '%s'", name)
+	}
+
+	hooks[name] = f
+}
+
+func registerFunc(name string, p Func) {
+	register(name, func(opts.Opts, env.Env) (hook, error) {
+		return p, nil
+	})
+}
+
+func GetList() []string {
+	var list []string
+
+	for k := range hooks {
+		list = append(list, k)
+	}
+
+	sort.Strings(list)
+
+	return list
+}
diff --git a/pkg/server/hook/split.go b/pkg/server/hook/split.go
new file mode 100644
index 0000000..75faf48
--- /dev/null
+++ b/pkg/server/hook/split.go
@@ -0,0 +1,57 @@
+package hook
+
+import (
+	"errors"
+	"strconv"
+	"tunnel/pkg/server/env"
+	"tunnel/pkg/server/opts"
+	"tunnel/pkg/server/queue"
+)
+
+const splitDefaultSize = 1024
+
+var errBadSize = errors.New("bad size value")
+
+type splitHook struct {
+	size int
+}
+
+func (h *splitHook) Send(rq, wq queue.Q) error {
+	for b := range rq {
+		var upto int
+
+		for n := 0; n < len(b); n = upto {
+			upto += h.size
+
+			if upto > len(b) {
+				upto = len(b)
+			}
+
+			wq <- b[n:upto]
+		}
+	}
+
+	return nil
+}
+
+func (h *splitHook) Open(env.Env) (interface{}, error) {
+	return h, nil
+}
+
+func newSplitHook(opts opts.Opts, env env.Env) (hook, error) {
+	size := splitDefaultSize
+
+	if s, ok := opts["size"]; ok {
+		var err error
+
+		if size, err = strconv.Atoi(s); err != nil || size <= 0 {
+			return nil, errBadSize
+		}
+	}
+
+	return &splitHook{size: size}, nil
+}
+
+func init() {
+	register("split", newSplitHook)
+}
diff --git a/pkg/server/hook/tee.go b/pkg/server/hook/tee.go
new file mode 100644
index 0000000..6591e33
--- /dev/null
+++ b/pkg/server/hook/tee.go
@@ -0,0 +1,120 @@
+package hook
+
+import (
+	"bytes"
+	"encoding/hex"
+	"fmt"
+	"os"
+	"path"
+	"sync"
+	"tunnel/pkg/server/env"
+	"tunnel/pkg/server/opts"
+	"tunnel/pkg/server/queue"
+)
+
+const teeDefaultFile = "/tmp/tunnel/dump"
+
+type tee struct {
+	f  *os.File
+	mu sync.Mutex
+	wg sync.WaitGroup
+}
+
+type teeHook struct {
+	file string
+}
+
+func (t *tee) dump(s string, p []byte) error {
+	var out bytes.Buffer
+
+	fmt.Fprintln(&out, s, len(p))
+
+	w := hex.Dumper(&out)
+	w.Write(p)
+	w.Close()
+
+	if _, err := t.f.Write(out.Bytes()); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (t *tee) Send(rq, wq queue.Q) error {
+	defer t.wg.Done()
+
+	for b := range rq {
+		t.dump(">", b)
+		wq <- b
+	}
+
+	return nil
+}
+
+func (t *tee) Recv(rq, wq queue.Q) error {
+	defer t.wg.Done()
+
+	for b := range rq {
+		t.dump("<", b)
+		wq <- b
+	}
+
+	return nil
+}
+
+func (h *teeHook) where(env env.Env) string {
+	if h.file != "" {
+		return h.file
+	}
+
+	if v := env.Eval("@{tunnel.@{tunnel}.tee.file}"); v != "" {
+		return v
+	}
+
+	if v, ok := env.Find("hook.tee.file"); ok {
+		return v
+	}
+
+	return teeDefaultFile
+}
+
+func (h *teeHook) Open(env env.Env) (interface{}, error) {
+	file := h.where(env)
+	dir := path.Dir(file)
+
+	if err := os.MkdirAll(dir, 0755); err != nil {
+		return nil, err
+	}
+
+	tid, sid := env.Get("tunnel"), env.Get("stream")
+	name := fmt.Sprintf("%s.%s.%s", file, tid, sid)
+
+	var t tee
+
+	if f, err := os.Create(name); err != nil {
+		return nil, err
+	} else {
+		t.f = f
+	}
+
+	t.wg.Add(2)
+
+	go func() {
+		t.wg.Wait()
+		t.f.Close()
+	}()
+
+	return &t, nil
+}
+
+func newTeeHook(opts opts.Opts, env env.Env) (hook, error) {
+	h := &teeHook{}
+	if file, ok := opts["file"]; ok {
+		h.file = file
+	}
+	return h, nil
+}
+
+func init() {
+	register("tee", newTeeHook)
+}
diff --git a/pkg/server/hook/zip.go b/pkg/server/hook/zip.go
new file mode 100644
index 0000000..61264c9
--- /dev/null
+++ b/pkg/server/hook/zip.go
@@ -0,0 +1,55 @@
+package hook
+
+import (
+	"compress/flate"
+	"io"
+	"tunnel/pkg/server/env"
+	"tunnel/pkg/server/opts"
+	"tunnel/pkg/server/queue"
+)
+
+type zipHook struct{}
+
+func (m zipHook) Send(rq, wq queue.Q) error {
+	w, err := flate.NewWriter(wq.Writer(), flate.BestCompression)
+	if err != nil {
+		return err
+	}
+
+	for b := range rq {
+		if _, err := w.Write(b); err != nil {
+			return err
+		}
+		if err := w.Flush(); err != nil {
+			return err
+		}
+	}
+
+	return w.Close()
+}
+
+func (m zipHook) Recv(rq, wq queue.Q) error {
+	r := flate.NewReader(rq.Reader())
+
+	// FIXME: not received ending due to ultimate conn.Close
+	if err := queue.IoCopy(r, wq.Writer()); err != nil {
+		if err == io.ErrUnexpectedEOF {
+			return nil
+		}
+		return err
+	}
+
+	return r.Close()
+}
+
+func (m zipHook) Open(env.Env) (interface{}, error) {
+	return m, nil
+}
+
+func newZipHook(opts.Opts, env.Env) (hook, error) {
+	return zipHook{}, nil
+}
+
+func init() {
+	register("zip", newZipHook)
+}