config builder

config/builder.go

@@ -0,0 +1,74 @@
+package config
+
+import (
+	"fmt"
+	"mbpl3p/proxy"
+	"mbpl3p/tcp"
+	"mbpl3p/tun"
+)
+
+// TODO: Delete this code as soon as an alternative is available
+type UselessMac bool
+
+func (UselessMac) CodeLength() int {
+	return 0
+}
+
+func (UselessMac) Generate([]byte) []byte {
+	return nil
+}
+
+func (u UselessMac) Verify([]byte, []byte) error {
+	return nil
+}
+
+func (c Configuration) Build() (*proxy.Proxy, error) {
+	p := proxy.Proxy{}
+	p.Generator = UselessMac(false)
+
+	for _, peer := range c.Peers {
+		p.Verifiers = append(p.Verifiers, UselessMac(true))
+
+		switch peer.Method {
+		case "TCP":
+			prod, cons, err := buildTcp(peer)
+			if err != nil {
+				return nil, err
+			}
+			p.Producers = append(p.Producers, prod)
+			p.Consumers = append(p.Consumers, cons)
+		}
+	}
+
+	ss, err := tun.NewTun("nc%d", 1500)
+	if err != nil {
+		return nil, err
+	}
+
+	p.Source = ss
+	p.Sink = ss
+
+	return &p, nil
+}
+
+func buildTcp(p Peer) (proxy.Producer, proxy.Consumer, error) {
+	if p.RemoteHost != "" {
+		f, err := tcp.NewFlow(
+			fmt.Sprintf("%s:", p.LocalHost),
+			fmt.Sprintf("%s:%d", p.RemoteHost, p.RemotePort),
+		)
+
+		if err != nil {
+			return nil, nil, err
+		}
+
+		return &f, &f, nil
+	}
+
+	r, err := tcp.NewReceiver(fmt.Sprintf("%s:%d", p.LocalHost, p.LocalPort))
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return r, r, nil
+}

config/config.go

@@ -10,16 +10,22 @@ type Configuration struct {
 }
 
 type Host struct {
-	PrivateKey string   `validate:"required"`
-	Methods    []string `validate:"min=1,dive,oneof=TCP"`
+	PrivateKey string `validate:"required"`
 }
 
 type Peer struct {
 	PublicKey string `validate:"required"`
 	Method    string `validate:"oneof=TCP"`
 
-	RemoteHost string `validate:"required,fqdn|ip"`
-	RemotePort uint   `validate:"required,max=65535"`
+	LocalHost string `validate:"omitempty,ip"`
+	LocalPort uint   `validate:"max=65535"`
+
+	RemoteHost string `validate:"required_with=RemotePort,omitempty,fqdn|ip"`
+	RemotePort uint   `validate:"required_with=RemoteHost,omitempty,max=65535"`
+
+	KeepAlive uint
+	Timeout   uint
+	RetryWait uint
 }
 
 func (c Configuration) Validate() error {

main.go

@@ -1,8 +1,10 @@
 package main
 
 import (
-	"fmt"
 	"mbpl3p/config"
+	"os"
+	"os/signal"
+	"syscall"
 )
 
 func main() {
@@ -11,5 +13,14 @@ func main() {
 		panic(err)
 	}
 
-	fmt.Printf("%+v\n", c)
+	p, err := c.Build()
+	if err != nil {
+		panic(err)
+	}
+	p.Start()
+
+	signals := make(chan os.Signal)
+	signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
+
+	<-signals
 }

proxy/mac.go

@@ -0,0 +1,11 @@
+package proxy
+
+type MacGenerator interface {
+	CodeLength() int
+	Generate([]byte) []byte
+}
+
+type MacVerifier interface {
+	CodeLength() int
+	Verify(data []byte, sum []byte) error
+}

proxy/proxy.go

@@ -1,6 +1,6 @@
 package proxy
 
-const ChecksumLength = 16
+import "fmt"
 
 type Producer interface {
 	IsAlive() bool
@@ -12,16 +12,6 @@ type Consumer interface {
 	Consume(Packet, MacGenerator) error
 }
 
-type MacGenerator interface {
-	CodeLength() int
-	Generate([]byte) []byte
-}
-
-type MacVerifier interface {
-	CodeLength() int
-	Verify(data []byte, sum []byte) error
-}
-
 type Source interface {
 	Source() (Packet, error)
 }
@@ -31,12 +21,70 @@ type Sink interface {
 }
 
 type Proxy struct {
-	Source Source
+	// TODO: Replace the slices with maps so they can be addressed more precisely
+	Source    Source
+	Generator MacGenerator
 	Consumers []Consumer
 
+	Verifiers []MacVerifier
 	Producers []Producer
-	Sink Sink
+	Sink      Sink
 
 	proxyChan chan Packet
 	sinkChan  chan Packet
 }
+
+func (p *Proxy) Start() {
+	p.proxyChan = make(chan Packet, 16)
+	p.sinkChan = make(chan Packet, 16)
+
+	go func() {
+		for {
+			if packet, err := p.Source.Source(); err != nil {
+				panic(err)
+				return
+			} else {
+				p.proxyChan <- packet
+			}
+		}
+	}()
+
+	for _, c := range p.Consumers {
+		c := c
+
+		go func() {
+			for c.IsAlive() {
+				if err := c.Consume(<-p.proxyChan, p.Generator); err != nil {
+					fmt.Println(err)
+					return
+				}
+			}
+		}()
+	}
+
+	for i, prod := range p.Producers {
+		i := i
+		prod := prod
+
+		go func() {
+			for prod.IsAlive() {
+				if packet, err := prod.Produce(p.Verifiers[i]); err != nil {
+					fmt.Println(err)
+					return
+				} else {
+					p.sinkChan <- packet
+				}
+			}
+		}()
+	}
+
+	go func() {
+		for {
+			packet := <-p.sinkChan
+			if err := p.Sink.Sink(packet); err != nil {
+				panic(err)
+				return
+			}
+		}
+	}()
+}

tcp/flow.go

@@ -6,38 +6,54 @@ import (
 	"io"
 	"mbpl3p/proxy"
 	"net"
-	"time"
 )
 
 var ErrNotEnoughBytes = errors.New("not enough bytes")
 
 type Flow struct {
-	Remote net.TCPAddr
+	Local  string
+	Remote string
 
-	conn   *net.TCPConn
+	conn *net.TCPConn
 }
 
-func NewFlow(remote net.TCPAddr) (Flow, error) {
-	f := Flow{Remote: remote}
-
-	var err error
-	f.conn, err = net.DialTCP("", nil, &f.Remote)
-
-	if err != nil {
-		return Flow{}, err
+func NewFlow(local, remote string) (Flow, error) {
+	f := Flow{
+		Local: local,
+		Remote: remote,
 	}
 
-	if err := f.conn.SetKeepAlive(true); err != nil {
-		return Flow{}, err
-	}
-
-	if err := f.conn.SetKeepAlivePeriod(30 * time.Second); err != nil {
+	if err := f.dial(); err != nil {
 		return Flow{}, err
 	}
 
 	return f, nil
 }
 
+func newFlowConn(conn *net.TCPConn) Flow {
+	return Flow{conn: conn}
+}
+
+func (f *Flow) dial() error {
+	localAddr, err := net.ResolveTCPAddr("tcp", f.Local)
+	if err != nil {
+		return err
+	}
+
+	remoteAddr, err := net.ResolveTCPAddr("tcp", f.Remote)
+	if err != nil {
+		return err
+	}
+
+	f.conn, err = net.DialTCP("tcp", localAddr, remoteAddr)
+
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 func (f *Flow) IsAlive() bool {
 	// TODO: Implement this
 	return true
@@ -45,7 +61,10 @@ func (f *Flow) IsAlive() bool {
 
 func (f *Flow) Consume(p proxy.Packet, g proxy.MacGenerator) error {
 	data := p.Marshal(g)
+	return f.consumeMarshalled(data)
+}
 
+func (f *Flow) consumeMarshalled(data []byte) error {
 	prefixedData := make([]byte, len(data)+4)
 	binary.LittleEndian.PutUint32(prefixedData, uint32(len(data)))
 	copy(prefixedData[4:], data)
@@ -55,20 +74,29 @@ func (f *Flow) Consume(p proxy.Packet, g proxy.MacGenerator) error {
 }
 
 func (f *Flow) Produce(v proxy.MacVerifier) (proxy.Packet, error) {
+	data, err := f.produceMarshalled()
+	if err != nil {
+		return proxy.Packet{}, err
+	}
+
+	return proxy.UnmarshalPacket(data, v)
+}
+
+func (f *Flow) produceMarshalled() ([]byte,error) {
 	lengthBytes := make([]byte, 4)
 	if n, err := io.LimitReader(f.conn, 4).Read(lengthBytes); err != nil {
-		return proxy.Packet{}, err
+		return nil, err
 	} else if n != 4 {
-		return proxy.Packet{}, ErrNotEnoughBytes
+		return nil, ErrNotEnoughBytes
 	}
 
 	length := binary.LittleEndian.Uint32(lengthBytes)
 	dataBytes := make([]byte, length)
 	if n, err := io.LimitReader(f.conn, int64(length)).Read(dataBytes); err != nil {
-		return proxy.Packet{}, err
+		return nil, err
 	} else if n != int(length) {
-		return proxy.Packet{}, ErrNotEnoughBytes
+		return nil, ErrNotEnoughBytes
 	}
 
-	return proxy.UnmarshalPacket(dataBytes, v)
-}
+	return dataBytes, nil
+}

tcp/receiver.go

@@ -1 +1,97 @@
 package tcp
+
+import (
+	"fmt"
+	"mbpl3p/proxy"
+	"mbpl3p/utils"
+	"net"
+)
+
+type Receiver struct {
+	flows map[int]Flow
+
+	inPackets  chan []byte
+	outPackets chan []byte
+}
+
+func NewReceiver(local string) (*Receiver, error) {
+	r := Receiver{}
+
+	laddr, err := net.ResolveTCPAddr("tcp", local)
+	if err != nil {
+		return nil, err
+	}
+
+	listener, err := net.ListenTCP("tcp", laddr)
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		for {
+			conn, err := listener.AcceptTCP()
+			if err != nil {
+				panic(err)
+			}
+
+			r.addFlow(newFlowConn(conn))
+		}
+	}()
+
+	return &r, nil
+}
+
+func (r *Receiver) IsAlive() bool {
+	return true
+}
+
+func (r *Receiver) Consume(p proxy.Packet, g proxy.MacGenerator) error {
+	r.inPackets <- p.Marshal(g)
+	return nil
+}
+
+func (r *Receiver) Produce(v proxy.MacVerifier) (proxy.Packet, error) {
+	return proxy.UnmarshalPacket(<-r.outPackets, v)
+}
+
+func (r *Receiver) addFlow(flow Flow) {
+	i := <-utils.NextId
+	r.flows[i] = flow
+
+	isAlive := make(chan bool)
+	notAlive := func() {
+		select {
+		case isAlive <- false:
+		default:
+		}
+	}
+
+	go func() {
+		for flow.IsAlive() {
+			d, err := flow.produceMarshalled()
+			if err != nil {
+				fmt.Println(err)
+				continue
+			}
+			r.inPackets <- d
+		}
+		notAlive()
+	}()
+
+	go func() {
+		for flow.IsAlive() {
+			d := <-r.outPackets
+			err := flow.consumeMarshalled(d)
+			if err != nil {
+				fmt.Println(err)
+				// continue
+			}
+		}
+		notAlive()
+	}()
+
+	go func() {
+		<-isAlive
+		delete(r.flows, i)
+	}()
+}

tun/tun.go

@@ -11,7 +11,9 @@ type SourceSink struct {
 	bufferSize int
 }
 
-func NewTun(namingScheme string, bufferSize int) (ss SourceSink, err error) {
+func NewTun(namingScheme string, bufferSize int) (ss *SourceSink, err error) {
+	ss = &SourceSink{}
+
 	ss.tun, err = taptun.NewTun(namingScheme)
 	if err != nil {
 		return

utils/utils.go

@@ -0,0 +1,11 @@
+package utils
+
+var NextId = make(chan int)
+
+func init() {
+	i := 0
+	for {
+		NextId <- i
+		i += 1
+	}
+}