commit 1b60da38c5bba3078a1801f4835216ad58d75b3f Author: Bel LaPointe Date: Thu May 2 10:08:11 2019 -0600 base diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..19fc60a --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +*.key +*.crt +*.pem +*.swp +*.swo +*.pub +cli-keys +fake-ssh +gcpkeys +*.json +*mnt +ssh-portable diff --git a/bop.go b/bop.go new file mode 100644 index 0000000..1b4ae23 --- /dev/null +++ b/bop.go @@ -0,0 +1,668 @@ +package encoder + +import ( + "bytes" + "compress/gzip" + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "encoding/base64" + "io" + "io/ioutil" + "local/encoder/pipe" + "local/encryptor" + "log" + "strconv" + + "github.com/golang/snappy" + "github.com/klauspost/reedsolomon" + "github.com/pierrec/lz4" +) + +var pipeBuffSize = int64(aes.BlockSize + 2) + +func SetPipeBuffSize(n int) { + if n < 50 { + n = 50 + } + pipeBuffSize = int64(n) +} + +type BOP3 interface { + Wrap3(chan []byte) chan []byte + Unwrap3(chan []byte) chan []byte + BOP +} + +type BOP2 interface { + Wrap2(io.Reader) io.Reader + Unwrap2(io.Reader) io.Reader + BOP +} + +type BOP interface { + Wrap([]byte) []byte + Unwrap([]byte) []byte +} + +type Base64 struct{} +type Zipper struct{} +type snapper struct{} +type symCryptor struct{ encryptor.Encryptor } +type asymCryptor struct{ encryptor.Encryptor } +type Salter struct{ n int } +type LZ4 struct{} +type AES struct { + Key []byte + keylen int + block cipher.Block +} +type PAR2 struct { + chunks int + pars int + enc reedsolomon.Encoder +} + +func (p *PAR2) fix() error { + if p.chunks == 0 { + p.chunks = 50 + } + if p.pars == 0 { + p.pars = 5 + } + if p.enc == nil { + e, err := reedsolomon.New(p.chunks, p.pars) + p.enc = e + return err + } + return nil +} +func (p *PAR2) Unwrap2(r io.Reader) io.Reader { + b, _ := ioutil.ReadAll(r) + b = p.Unwrap(b) + return bytes.NewReader(b) +} +func (p *PAR2) Wrap2(r io.Reader) io.Reader { + b, _ := ioutil.ReadAll(r) + b = p.Wrap(b) + return bytes.NewReader(b) +} +func (p *PAR2) Wrap(b []byte) []byte { + if p.fix() != nil { + return b + } + split, err := p.enc.Split(b) + if err != nil { + return b + } + if err := p.enc.Encode(split); err != nil { + return b + } + if ok, err := p.enc.Verify(split); err != nil || !ok { + return b + } + l := strconv.Itoa(len(b)) + l = string(byte(len(l))) + l + b = []byte(l) + for i := range split { + b = append(b, split[i]...) + } + return b +} +func (p *PAR2) Unwrap(b []byte) []byte { + if p.fix() != nil { + return b + } + l := int(b[0]) + o := b[1 : l+1] + c := b[l+1:] + m, err := strconv.Atoi(string(o)) + if err != nil { + return b + } + split := make([][]byte, p.chunks+p.pars) + chunklen := len(c) / (p.chunks + p.pars) + for i := range split { + split[i] = c[i*chunklen : (i+1)*chunklen] + } + if err := p.enc.Reconstruct(split); err != nil { + return b + } + d := []byte{} + for i := 0; i < p.chunks; i++ { + d = append(d, split[i]...) + } + return d[:m] +} + +func (z *Zipper) Wrap3(b chan []byte) chan []byte { + out := make(chan []byte) + go func() { + defer close(out) + pipeR, pipeW := io.Pipe() + go func() { + zw := gzip.NewWriter(pipeW) + defer pipeW.Close() + defer zw.Close() + for input := range b { + if _, err := zw.Write(input); err != nil { + panic(err) + } + } + }() + subbuff := make([]byte, pipeBuffSize) + for { + n, err := pipeR.Read(subbuff) + if err != nil && err != io.EOF { + panic(err) + } else if n > 0 { + out <- copyBSlice(subbuff[:n]) + } + if err == io.EOF { + break + } + } + }() + return out +} +func (z *Zipper) Unwrap3(b chan []byte) chan []byte { + out := make(chan []byte) + go func() { + pipeR, pipeW := io.Pipe() + defer close(out) + go func() { + defer pipeW.Close() + for input := range b { + if _, err := pipeW.Write(input); err != nil { + panic(err) + } + } + }() + zr, err := gzip.NewReader(pipeR) + if err != nil { + panic(err) + } + buff := make([]byte, pipeBuffSize) + for { + n, err := zr.Read(buff) + if err != nil && err != io.EOF { + panic(err) + } + if n > 0 { + out <- copyBSlice(buff[:n]) + } + if err == io.EOF { + break + } + } + }() + return out +} + +func (z *Zipper) Wrap2(b io.Reader) io.Reader { + pipe := pipe.New() + pipe.SetCap(pipeBuffSize) + zw := gzip.NewWriter(pipe) + go func() { + defer pipe.Close() + defer zw.Close() + for { + m, err := io.CopyN(zw, b, pipeBuffSize) + if err != nil && err != io.EOF { + panic(err) + } + if m > 0 { + zw.Flush() + } + if err == io.EOF { + return + } + } + /* + internalBuffer := make([]byte, pipeBuffSize/2+bytes.MinRead) + if _, err := io.CopyBuffer(zw, b, internalBuffer); err != nil { + panic(err) + } + */ + }() + return pipe +} +func (z *Zipper) Unwrap2(b io.Reader) io.Reader { + pipe := pipe.New() + pipe.SetCap(pipeBuffSize) + go func() { + defer pipe.Close() + zr, err := gzip.NewReader(b) + if err != nil && err != io.EOF { + panic(err) + } else if err == nil { + if _, err := io.Copy(pipe, zr); err != nil { + panic(err) + } + defer zr.Close() + } + }() + return pipe +} + +func (z *Zipper) Wrap(b []byte) []byte { + var buf bytes.Buffer + zw := gzip.NewWriter(&buf) + if _, err := zw.Write(b); err != nil { + return b + } + zw.Close() + return buf.Bytes() +} + +func (z *Zipper) Unwrap(b []byte) []byte { + buf := bytes.NewBuffer(b) + zw, err := gzip.NewReader(buf) + if err != nil { + return b + } + decompressed, err := ioutil.ReadAll(zw) + if err != nil { + return b + } + return decompressed +} + +func (s *Salter) Unwrap3(b chan []byte) chan []byte { + out := make(chan []byte) + go func() { + defer close(out) + first := <-b + saltLen := int(first[0]) + for len(first) < saltLen+1 { + first = append(first, <-b...) + } + first = first[saltLen+1:] + out <- copyBSlice(first) + for input := range b { + out <- copyBSlice(input) + } + }() + return out +} + +func (s *Salter) Wrap3(b chan []byte) chan []byte { + out := make(chan []byte) + salt := s.makeSalt() + go func() { + defer close(out) + out <- copyBSlice(salt) + for input := range b { + out <- copyBSlice(input) + } + }() + return out +} + +func (s *Salter) makeSalt() []byte { + n := 21 + if s.n > 0 { + n = s.n + } + salt := make([]byte, n) + _, err := rand.Read(salt) + if err != nil { + return nil + } + return append([]byte{byte(len(salt))}, salt...) +} + +func (s *Salter) Wrap2(b io.Reader) io.Reader { + salt := s.makeSalt() + + pipe := pipe.New() + pipe.SetCap(pipeBuffSize) + go func() { + defer pipe.Close() + if _, err := pipe.Write(salt); err != nil { + panic(err) + } + tmpbuff := make([]byte, pipeBuffSize) + if _, err := io.CopyBuffer(pipe, b, tmpbuff); err != nil { + panic(err) + } + }() + return pipe +} + +func (s *Salter) Unwrap2(b io.Reader) io.Reader { + buffer := bytes.NewBuffer(nil) + if _, err := io.CopyN(buffer, b, 1); err != nil { + panic(err) + } + l := int(buffer.Bytes()[0]) + if _, err := io.CopyN(buffer, b, int64(l)); err != nil { + panic(err) + } + pipe := pipe.New() + pipe.SetCap(pipeBuffSize) + go func() { + defer pipe.Close() + if _, err := io.Copy(pipe, b); err != nil && err != io.EOF { + panic(err) + } + }() + return pipe +} + +func (s *Salter) Wrap(b []byte) []byte { + salt := s.makeSalt() + + return append(salt, b...) +} + +func (s *Salter) Unwrap(b []byte) []byte { + if len(b) < 1 { + return b + } + l := int(b[0]) + if l > len(b)+1 { + return b + } + return b[1+l:] +} + +func (sc *symCryptor) Wrap(b []byte) []byte { + return []byte(sc.Encrypt(string(b))) +} + +func (sc *symCryptor) Unwrap(b []byte) []byte { + return []byte(sc.Decrypt(string(b))) +} + +func (sc *asymCryptor) Wrap(b []byte) []byte { + return []byte(sc.Encrypt(string(b))) +} + +func (sc *asymCryptor) Unwrap(b []byte) []byte { + return []byte(sc.Decrypt(string(b))) +} + +func (s *snapper) Wrap(b []byte) []byte { + return snappy.Encode(nil, b) +} + +func (s *snapper) Unwrap(b []byte) []byte { + d, err := snappy.Decode(nil, b) + if err != nil { + return b + } + return d +} + +func (s *AES) Wrap3(b chan []byte) chan []byte { + s.fix() + out := make(chan []byte) + + go func() { + defer close(out) + + // IV + iv := make([]byte, aes.BlockSize) + if n, err := io.ReadFull(rand.Reader, iv); err != nil || n != len(iv) { + panic(err) + } + out <- copyBSlice(iv) + + // cipher + ttl := int64(0) + stream := cipher.NewCFBEncrypter(s.block, iv) + for input := range b { + ttl += int64(len(input)) + stream.XORKeyStream(input, input) + if len(input) > 0 { + out <- copyBSlice(input) + } + } + + // padding + out <- copyBSlice(s.getPadding(ttl)) + }() + + return out +} + +func (s *AES) Wrap2(b io.Reader) io.Reader { + s.fix() + pipe := pipe.New() + pipe.SetCap(pipeBuffSize) + + go func() { + defer pipe.Close() + + // IV + iv := make([]byte, aes.BlockSize) + if n, err := io.ReadFull(rand.Reader, iv); err != nil || n != len(iv) { + panic(err) + } + pipe.Write(iv) + + // cipher + ttl := int64(0) + stream := cipher.NewCFBEncrypter(s.block, iv) + buff := make([]byte, pipeBuffSize) + for n, err := b.Read(buff); err == nil || err == io.EOF; n, err = b.Read(buff) { + ttl += int64(n) + stream.XORKeyStream(buff[:n], buff[:n]) + if m, err := pipe.Write(buff[:n]); err != nil || m != n { + panic(err) + } + if err == io.EOF { + break + } + } + + // padding + pipe.Write(s.getPadding(ttl)) + }() + + return pipe +} + +func (s *AES) Unwrap2(b io.Reader) io.Reader { + s.fix() + pipe := pipe.New() + pipe.SetCap(pipeBuffSize) + + go func() { + defer pipe.Close() + + // IV + var iv []byte + add := make([]byte, pipeBuffSize) + for len(iv) < aes.BlockSize { + n, err := b.Read(add) + if err != nil && err != io.EOF { + panic(err) + } + iv = append(iv, add[:n]...) + } + last := iv[aes.BlockSize:] + iv = iv[:aes.BlockSize] + + // cipher + var n int + var err error + stream := cipher.NewCFBDecrypter(s.block, iv) + for n, err = b.Read(add); err == nil || err == io.EOF; n, err = b.Read(add) { + last = append(last, add[:n]...) + stream.XORKeyStream(last[:len(last)-s.needModZero()], last[:len(last)-s.needModZero()]) + pipe.Write(last[:len(last)-s.needModZero()]) + last = last[len(last)-s.needModZero():] + if err == io.EOF { + break + } + } + if err != nil && err != io.EOF { + panic(err) + } + // padding + trim := int(last[len(last)-1]) + last = last[:len(last)-trim] + stream.XORKeyStream(last, last) + pipe.Write(last) + }() + + return pipe +} + +func (s *AES) Unwrap3(b chan []byte) chan []byte { + s.fix() + out := make(chan []byte) + + go func() { + defer close(out) + + // IV + var lastTwo [2][]byte + var iv []byte + for len(iv) < aes.BlockSize { + add := <-b + if len(add) >= aes.BlockSize-len(iv) { + isIV := aes.BlockSize - len(iv) + lastTwo[0] = add[isIV:] + add = add[:isIV] + } + iv = append(iv, add...) + } + + // cipher + stream := cipher.NewCFBDecrypter(s.block, iv) + for input := range b { + stream.XORKeyStream(lastTwo[1], lastTwo[1]) + if len(lastTwo[1]) > 0 { + out <- copyBSlice(lastTwo[1]) + } + lastTwo[1] = lastTwo[0] + lastTwo[0] = input + } + last := append(lastTwo[1], lastTwo[0]...) + // padding + trim := int(last[len(last)-1]) + last = last[:len(last)-trim] + stream.XORKeyStream(last, last) + if len(last) > 0 { + out <- copyBSlice(last) + } + }() + + return out +} + +func (s *AES) Wrap(b []byte) []byte { + s.fix() + + c := s.pad(b) + + d := make([]byte, aes.BlockSize+len(c)) + iv := d[:aes.BlockSize] + if _, err := io.ReadFull(rand.Reader, iv); err != nil { + return nil + } + + stream := cipher.NewCFBEncrypter(s.block, iv) + stream.XORKeyStream(d[aes.BlockSize:], c) + + return d +} + +func (s *AES) Unwrap(b []byte) []byte { + s.fix() + + iv := b[:aes.BlockSize] + b = b[aes.BlockSize:] + + stream := cipher.NewCFBDecrypter(s.block, iv) + stream.XORKeyStream(b, b) + + return s.unpad(b) +} +func (s *AES) fix() { + if s.keylen < 1 { + s.keylen = 32 + } + if len(s.Key) != s.keylen { + s.Key = bytes.Repeat(append(s.Key, 0), s.keylen)[:s.keylen] + } + if s.block == nil { + s.block, _ = aes.NewCipher(s.Key) + } +} + +func (s *AES) unpad(b []byte) []byte { + s.fix() + if len(b) < 1 || len(b) < int(b[len(b)-1]) { + return b + } + return b[:len(b)-int(b[len(b)-1])] +} + +func (s *AES) pad(b []byte) []byte { + s.fix() + return append(b, s.getPadding(int64(len(b)))...) +} + +func (s *AES) getPadding(l int64) []byte { + padding := int(int64(s.needModZero()) - l%int64(s.needModZero())) + return bytes.Repeat([]byte{byte(padding)}, padding) +} + +func (s *AES) needModZero() int { + return aes.BlockSize +} + +func (l *LZ4) Wrap(b []byte) []byte { + var buf bytes.Buffer + w := lz4.NewWriter(&buf) + if _, err := io.Copy(w, bytes.NewBuffer(b)); err != nil { + panic(err) + } + if err := w.Close(); err != nil { + panic(err) + } + return buf.Bytes() +} + +func (l *LZ4) Unwrap(b []byte) []byte { + zr := lz4.NewReader(bytes.NewBuffer(b)) + b, err := ioutil.ReadAll(zr) + if err != nil { + panic(err) + } + return b +} + +func copyBSlice(b []byte) []byte { + return append([]byte{}, b...) +} + +func screenReader(b io.Reader) io.Reader { + return b + c, err := ioutil.ReadAll(b) + if err != nil { + panic(err) + } + log.Printf("screened %v", len(c)) + if len(c) > 20 { + log.Printf("%s...%s", c[:5], c[len(c)-5:]) + } else if len(c) > 0 { + log.Printf("%s", c) + } + return bytes.NewReader(c) +} + +func (b64 *Base64) Wrap(b []byte) []byte { + return []byte(base64.StdEncoding.EncodeToString(b)) +} +func (b64 *Base64) Unwrap(b []byte) []byte { + b, err := base64.StdEncoding.DecodeString(string(b)) + if err != nil { + return nil + } + return b +} diff --git a/bop2_test.go b/bop2_test.go new file mode 100644 index 0000000..1dbb751 --- /dev/null +++ b/bop2_test.go @@ -0,0 +1,61 @@ +package encoder + +import ( + "io/ioutil" + "strings" + "testing" +) + +func Test_BOP2(t *testing.T) { + cases := []struct { + bop BOP2 + streamLength int + bufferCapacity int + }{ + { + bop: &Salter{}, + streamLength: 50000, + bufferCapacity: 100, + }, + { + bop: &Salter{}, + streamLength: 50, + bufferCapacity: 40, + }, + { + bop: &AES{Key: []byte("key")}, + streamLength: 50000, + bufferCapacity: 100, + }, + { + bop: &AES{Key: []byte("key")}, + streamLength: 50, + bufferCapacity: 40, + }, + { + bop: &Zipper{}, + streamLength: 50000, + bufferCapacity: 100, + }, + { + bop: &Zipper{}, + streamLength: 50, + bufferCapacity: 40, + }, + } + + for _, c := range cases { + SetPipeBuffSize(c.bufferCapacity) + inputString := strings.Repeat("hello world! ", c.streamLength)[:c.streamLength] + input := strings.NewReader(inputString) + wrappedReader := c.bop.Wrap2(input) + unwrappedReader := c.bop.Unwrap2(wrappedReader) + out, err := ioutil.ReadAll(unwrappedReader) + if err != nil { + t.Fatalf("failed to read from unwrapped: %v", err) + } else if string(out) != inputString { + t.Errorf("wrong output: wanted (%v), got (%v)", len(inputString), len(out)) + } + t.Logf("%T : %v => ? => %v", c.bop, len(inputString), len(out)) + } +} diff --git a/bop3_test.go b/bop3_test.go new file mode 100644 index 0000000..7cb5aa5 --- /dev/null +++ b/bop3_test.go @@ -0,0 +1,75 @@ +package encoder + +import ( + "strings" + "testing" +) + +func Test_BOP3(t *testing.T) { + was := pipeBuffSize + defer func() { + SetPipeBuffSize(int(was)) + }() + cases := []struct { + bop BOP3 + sz int + }{ + { + sz: 500, + bop: &AES{Key: []byte("hi")}, + }, + { + sz: 10, + bop: &AES{Key: []byte("hi")}, + }, + { + sz: 500, + bop: &Zipper{}, + }, + { + sz: 10, + bop: &Zipper{}, + }, + { + sz: 500, + bop: &Salter{}, + }, + { + sz: 10, + bop: &Salter{}, + }, + } + + for _, c := range cases { + SetPipeBuffSize(c.sz) + input := strings.Repeat("input", 5) + inch := make(chan []byte) + wrapch := c.bop.Wrap3(inch) + go func() { + for _, ch := range input { + inch <- []byte{byte(ch)} + } + close(inch) + }() + wrapped := "" + for o := range wrapch { + wrapped += string(o) + } + if len(wrapped) < 1 { + t.Errorf("%T : %s => 0 length", c, input) + continue + } + inunwrapch := make(chan []byte) + unwrapch := c.bop.Unwrap3(inunwrapch) + inunwrapch <- []byte(wrapped) + close(inunwrapch) + out := "" + for o := range unwrapch { + out += string(o) + } + if input != out { + t.Errorf("wrap3 failed for %T", c) + } + t.Logf("%T : %v : %s => (%v) => %s", c.bop, c.sz, input, len(wrapped), out) + } +} diff --git a/bop_test.go b/bop_test.go new file mode 100644 index 0000000..eafe986 --- /dev/null +++ b/bop_test.go @@ -0,0 +1,112 @@ +package encoder + +import ( + "fmt" + "io/ioutil" + "local/encryptor" + "os" + "path" + "strings" + "testing" +) + +func benchmarkBopOp(bop BOP, op int, l int, b *testing.B) { + dir, err := os.Getwd() + if err != nil { + b.Fatalf("cannot get wd: %v", err) + } + input, err := ioutil.ReadFile(path.Join(dir, "bop.go")) + if err != nil { + b.Fatalf("cannot read bop.go: %v", err) + } + for len(input) < l { + input = append(input, input...) + } + if len(input) > l { + input = input[:l] + } + switch op { + case 0: + case 1: + input = bop.Wrap(input) + } + b.N = 4 + b.Run(fmt.Sprintf("%T-%v-%v", bop, op, len(input)), func(b *testing.B) { + for i := 0; i < b.N; i++ { + switch op { + case 0: + bop.Wrap([]byte(string(input))) + case 1: + bop.Unwrap([]byte(string(input))) + } + } + }) +} + +func Test_BOPs_Mem(t *testing.T) { + cases := []struct { + bop BOP + }{ + {bop: &AES{Key: []byte("hello")}}, + {bop: &PAR2{}}, + {bop: &LZ4{}}, + {bop: &Zipper{}}, + } + + base := 50000000 + scale := 10 + for _, c := range cases { + memoryRatioByOp := [4]float64{} + for op := 0; op < 4; op++ { + l := base + if op > 1 { + l *= scale + } + //name := fmt.Sprintf("%T-%d-%d", c.bop, op%2, l) + result := testing.Benchmark(func(b *testing.B) { + b.ReportAllocs() + b.SetBytes(1) + benchmarkBopOp(c.bop, op%2, l, b) + }) + if result.MemBytes > 0 { + memoryRatioByOp[op] = float64(result.MemBytes) / float64(l) + } + //t.Logf("%s: %v", name, memoryRatioByOp[op]) + } + for i := 0; i < 2; i++ { + t.Logf("%T-%d: %v -> %v (%v MB)", c.bop, i, memoryRatioByOp[i], memoryRatioByOp[i+2], memoryRatioByOp[i+2]*float64(scale)*float64(base)/1000/1000) + } + } +} + +func Test_BOPs(t *testing.T) { + cases := []struct { + name string + bop BOP + }{ + {name: "zipper", bop: &Zipper{}}, + {name: "snapper", bop: &snapper{}}, + {name: "symC", bop: &symCryptor{encryptor.NewEncryptor("", "")}}, + {name: "asymC", bop: &asymCryptor{encryptor.NewEncryptor("", "")}}, + {name: "salter", bop: &Salter{}}, + {name: "lz4", bop: &LZ4{}}, + {name: "aes", bop: &AES{Key: []byte("hello")}}, + {name: "par2", bop: &PAR2{}}, + {name: "b64", bop: &Base64{}}, + } + + for _, c := range cases { + input := []byte(strings.Repeat("Hello world", 100)) + encoded := c.bop.Wrap(input) + decoded := c.bop.Unwrap(encoded) + if string(input) != string(decoded) { + t.Errorf("BOP %v failed: len(inp)=%v, len(enc)=%v, len(dec)=%v", c.name, len(input), len(encoded), len(decoded)) + } + } +} + +func Test_AES_Padding(t *testing.T) { + for i := 0; i < 100; i++ { + t.Logf("%v => %v", i, len((&AES{}).getPadding(int64(i)))) + } +} diff --git a/config.go b/config.go new file mode 100644 index 0000000..7755a06 --- /dev/null +++ b/config.go @@ -0,0 +1,7 @@ +package encoder + +type Config struct { + SymKey string + SymKeyOnly bool + KeyPath string +} diff --git a/encoder.go b/encoder.go new file mode 100644 index 0000000..c9aad98 --- /dev/null +++ b/encoder.go @@ -0,0 +1,252 @@ +package encoder + +import ( + "errors" + "io" + "local/encryptor" + "os" + "path" +) + +type Encoder struct { + bops []BOP +} + +func New(config *Config, bops ...BOP) (*Encoder, error) { + if len(bops) != 0 { + return &Encoder{bops: bops}, nil + } + + bops = []BOP{ + &Zipper{}, + &Salter{}, + } + + if config != nil { + if config.SymKey != "" { + bops = append(bops, &AES{Key: []byte(config.SymKey)}) + } + + if !config.SymKeyOnly { + var encr encryptor.Encryptor + if path.Base(config.KeyPath) == "env" { + pub := os.Getenv("PUB") + pri := os.Getenv("PRI") + if pub == "" || pri == "" { + return nil, errors.New("PUB and PRI cannot be empty") + } + encr = encryptor.NewEncryptor(pri, pub, false) + } else if config.KeyPath != "" { + encr = encryptor.NewEncryptor("", "", false) + priKeyPath := path.Join(config.KeyPath, "key.pri") + pubKeyPath := path.Join(config.KeyPath, "key.pub") + defer encr.ToFiles(priKeyPath, pubKeyPath) + if err := encr.FromFiles(priKeyPath, pubKeyPath); err != nil { + return nil, err + } + } else { + encr = encryptor.NewEncryptor("", "", false) + } + bops = append(bops, &asymCryptor{Encryptor: encr}) + } + } + + bops = append(bops, &Zipper{}) + return &Encoder{ + bops: bops, + }, nil +} + +func (e *Encoder) Encode2(r io.Reader) (io.Reader, error) { + pipeR, pipeW := io.Pipe() + go func() { + defer pipeW.Close() + for i := 0; i < len(e.bops); i++ { + r = e.bops[i].(BOP2).Wrap2(r) + } + if _, err := io.Copy(pipeW, r); err != nil { + panic(err) + } + }() + return pipeR, nil +} + +func (e *Encoder) Decode2(r io.Reader) (io.Reader, error) { + pipeR, pipeW := io.Pipe() + go func() { + defer pipeW.Close() + r = screenReader(r) + for i := len(e.bops) - 1; i >= 0; i-- { + r = e.bops[i].(BOP2).Unwrap2(r) + r = screenReader(r) + } + if _, err := io.Copy(pipeW, r); err != nil { + panic(err) + } + }() + return pipeR, nil +} + +func (e *Encoder) Encode3(r io.Reader) (io.Reader, error) { + pipeR, pipeW := io.Pipe() + + bopCh := make([]chan []byte, len(e.bops)+1) + inputChIndex := len(bopCh) - 1 + bopCh[inputChIndex] = make(chan []byte) + //[0]=bops[0]([n+1]) + //[1]=bops[1]([0]) + //... + //[n]=bops[n]([n-1]) + //[n+1]=firstin + for i := range e.bops { + bopCh[i] = e.bops[i].(BOP3).Wrap3(bopCh[(i-1+len(bopCh))%len(bopCh)]) + } + go func() { + defer pipeW.Close() + for input := range bopCh[inputChIndex-1] { + if _, err := pipeW.Write(input); err != nil { + panic(err) + } + } + }() + go func() { + defer close(bopCh[inputChIndex]) + buff := make([]byte, 64000) + for { + n, err := r.Read(buff) + if err != nil && err != io.EOF { + panic(err) + } + if n > 0 { + bopCh[inputChIndex] <- copyBSlice(buff[:n]) + } + if err == io.EOF { + break + } + } + }() + return pipeR, nil +} + +func (e *Encoder) Decode3(r io.Reader) (io.Reader, error) { + pipeR, pipeW := io.Pipe() + + bopCh := make([]chan []byte, len(e.bops)+1) + inputChIndex := len(bopCh) - 1 + bopCh[inputChIndex] = make(chan []byte) + //[0]=bops[0]([1]) + //[1]=bops[1]([2]) + //... + //[n]=bops[n]([n+1]) + //[n+1]=firstin + for i := len(e.bops) - 1; i >= 0; i-- { + bopCh[i] = e.bops[i].(BOP3).Unwrap3(bopCh[i+1]) + } + go func() { + defer pipeW.Close() + for input := range bopCh[0] { + if _, err := pipeW.Write(input); err != nil { + panic(err) + } + } + }() + go func() { + defer close(bopCh[inputChIndex]) + buff := make([]byte, 64000) + for { + n, err := r.Read(buff) + if err != nil && err != io.EOF { + panic(err) + } + if n > 0 { + bopCh[inputChIndex] <- copyBSlice(buff[:n]) + } + if err == io.EOF { + break + } + } + }() + return pipeR, nil +} + +func (e *Encoder) Encode(payload []byte, metadata ...[]byte) ([]byte, error) { + if len(payload) == 0 { + return nil, nil + } + + for i := range e.bops { + payload = e.bops[i].Wrap(payload) + if len(payload) == 0 { + return nil, errors.New("cannot encode payload") + } + } + + return e.WriteMetadata(payload, metadata) +} + +func (e *Encoder) Decode(payload []byte) ([]byte, error) { + metalen, err := e.metadataLength(payload) + if err != nil { + return nil, err + } + payload = payload[metalen:] + + if len(payload) == 0 { + return nil, nil + } + + for i := len(e.bops); i > 0; i-- { + payload = e.bops[i-1].Unwrap(payload) + if len(payload) == 0 { + return nil, errors.New("cannot decode payload") + } + } + + return payload, nil +} + +func (e *Encoder) WriteMetadata(payload []byte, metadata [][]byte) ([]byte, error) { + meta := []byte{byte(len(metadata))} + for i := range metadata { + meta = append(meta, byte(len(metadata[i]))) + meta = append(meta, metadata[i]...) + } + payload = append(meta, payload...) + return payload, nil +} + +func (e *Encoder) ReadMetadata(payload []byte) ([][]byte, error) { + metadata := [][]byte{} + if len(payload) < 2 { + return nil, nil + } + metalen := int(payload[0]) + payload = payload[1:] + for i := 0; i < metalen; i++ { + if len(payload) < 1 { + return nil, nil + } + j := payload[0] + if len(payload) < 1+int(j) { + return nil, nil + } + metadata = append(metadata, payload[1:1+int(j)]) + payload = payload[1+int(j):] + } + return metadata, nil +} + +func (e *Encoder) metadataLength(payload []byte) (int, error) { + metadata, err := e.ReadMetadata(payload) + if err != nil { + return 0, err + } + n := 1 + for i := range metadata { + n += 1 + len(metadata[i]) + } + if len(payload) < n { + return 0, nil + } + return n, nil +} diff --git a/encoder_test.go b/encoder_test.go new file mode 100644 index 0000000..8360279 --- /dev/null +++ b/encoder_test.go @@ -0,0 +1,421 @@ +package encoder + +import ( + "bytes" + "crypto/aes" + "fmt" + "io" + "io/ioutil" + "local/encryptor" + "log" + "strings" + "testing" +) + +func Test_Encoder2(t *testing.T) { + cases := []struct { + bops []BOP + streamLength int + bufferCapacity int + }{ + { + bops: []BOP{ + &AES{Key: []byte("key")}, + }, + streamLength: 75, + bufferCapacity: 25, + }, + { + bops: []BOP{ + &Salter{}, + }, + streamLength: 75, + bufferCapacity: 25, + }, + { + bops: []BOP{ + &Zipper{}, + }, + streamLength: 75, + bufferCapacity: 25, + }, + { + bops: []BOP{ + &AES{Key: []byte("key")}, + }, + streamLength: 7500, + bufferCapacity: 250, + }, + { + bops: []BOP{ + &Salter{}, + }, + streamLength: 7500, + bufferCapacity: 250, + }, + { + bops: []BOP{ + &Zipper{}, + }, + streamLength: 7500, + bufferCapacity: 250, + }, + { + bops: []BOP{ + &Zipper{}, + &Salter{}, + &AES{Key: []byte("key")}, + }, + streamLength: 75, + bufferCapacity: 50, + }, + { + bops: []BOP{ + &Zipper{}, + &Salter{}, + &AES{Key: []byte("key")}, + &Zipper{}, + }, + streamLength: 0, + bufferCapacity: 250, + }, + { + bops: []BOP{ + &Zipper{}, + &Salter{}, + &AES{Key: []byte("key")}, + &Zipper{}, + }, + streamLength: 76 * 1000, + bufferCapacity: 250, + }, + { + bops: []BOP{ + &Zipper{}, + &Salter{}, + &AES{Key: []byte("key")}, + &Zipper{}, + }, + streamLength: 7500, + bufferCapacity: 250, + }, + { + bops: []BOP{ + &Zipper{}, + &Salter{}, + &AES{Key: []byte("key")}, + &Zipper{}, + }, + streamLength: 75, + bufferCapacity: 25, + }, + } + + for _, c := range cases { + log.Printf("test cap %v, input length %v, bop length %v", c.bufferCapacity, c.streamLength, len(c.bops)) + SetPipeBuffSize(c.bufferCapacity) + enc, _ := New(nil, c.bops...) + input := strings.Repeat("hello world", c.streamLength)[:c.streamLength] + inputReader := strings.NewReader(input) + encodedReader, err := enc.Encode2(inputReader) + if err != nil { + t.Fatalf("cannot encode2: %v", err) + } + encoded, err := ioutil.ReadAll(encodedReader) + if err != nil { + t.Fatalf("cannot read encdoed: %v", err) + } + encodedReader = bytes.NewReader(encoded) + encodedBytes, err := ioutil.ReadAll(encodedReader) + if err != nil { + t.Fatalf("cannot read encoded all: %v", err) + } + log.Printf("%v -> %v -> ?", c.streamLength, len(encodedBytes)) + decodedReader, err := enc.Decode2(bytes.NewBuffer(encodedBytes)) + if err != nil { + t.Fatalf("cannot decode2: %v", err) + } + out, err := ioutil.ReadAll(decodedReader) + if err != nil { + t.Fatalf("cannot gather decoded: %v", err) + } + if string(out) != input { + t.Errorf("wrong encdoed-decoded: got %v, wanted %v", len(out), len(input)) + } + } +} + +func Test_Encoder3_Enc(t *testing.T) { + was := pipeBuffSize + defer func() { + SetPipeBuffSize(int(was)) + }() + SetPipeBuffSize(aes.BlockSize + 2) + cases := []struct { + bops []BOP + }{ + {bops: []BOP{&Salter{}}}, + {bops: []BOP{&Zipper{}}}, + {bops: []BOP{&Zipper{}, &Salter{}}}, + {bops: []BOP{&Salter{}, &Zipper{}}}, + {bops: []BOP{&AES{Key: []byte("key")}}}, + } + + for _, c := range cases { + title := "---" + for i := range c.bops { + title += fmt.Sprintf(" %T", c.bops[i]) + } + enc, err := New(nil, c.bops...) + if err != nil { + t.Fatalf("cannot create new encoder: %v", err) + } + input := strings.Repeat("hello world", int(pipeBuffSize*2))[:pipeBuffSize*2] + inputReader := bytes.NewBuffer([]byte(input)) + encodeReader, err := enc.Encode3(inputReader) + if err != nil { + t.Fatalf("cannot get encode reader: %v", err) + } + out, err := ioutil.ReadAll(encodeReader) + if err != nil { + t.Fatalf("cannot read from decode reader: %v", err) + } + t.Logf("%s : %s => %s", title, input, out) + } +} +func Test_Encoder3_All(t *testing.T) { + was := pipeBuffSize + defer func() { + SetPipeBuffSize(int(was)) + }() + SetPipeBuffSize(aes.BlockSize + 2) + cases := []struct { + bops []BOP + }{ + {bops: []BOP{&Salter{}}}, + {bops: []BOP{&Zipper{}}}, + {bops: []BOP{&AES{Key: []byte("key")}}}, + {bops: []BOP{&Zipper{}, &Salter{}}}, + {bops: []BOP{&Salter{}, &Zipper{}}}, + {bops: []BOP{&Zipper{}, &Salter{}, &AES{Key: []byte("key")}, &Zipper{}}}, + } + + for _, c := range cases { + title := "///" + for i := range c.bops { + title += fmt.Sprintf(" %T", c.bops[i]) + } + enc, err := New(nil, c.bops...) + if err != nil { + t.Fatalf("cannot create new encoder: %v", err) + } + input := strings.Repeat("hello world", int(pipeBuffSize*2))[:pipeBuffSize*2] + inputReader := bytes.NewBuffer([]byte(input)) + encodeReader, err := enc.Encode3(inputReader) + if err != nil { + t.Fatalf("cannot get encode reader: %v", err) + } + var decodeReader io.Reader + if false { + encodedString, err := ioutil.ReadAll(encodeReader) + if err != nil { + t.Fatalf("cannot gather encoded: %v", err) + } + decodeReader, err = enc.Decode3(bytes.NewBuffer(encodedString)) + } else { + decodeReader, err = enc.Decode3(encodeReader) + } + if err != nil { + t.Fatalf("cannot get decode reader: %v", err) + } + out, err := ioutil.ReadAll(decodeReader) + if err != nil { + t.Fatalf("cannot read from decode reader: %v", err) + } + if string(out) != input { + t.Errorf("failed test with %v bops", len(c.bops)) + } + t.Logf("%v : %s => %s", c.bops, input, out) + } +} + +func Test_Encoder(t *testing.T) { + input := "My secret secret" + conf := &Config{ + SymKey: "letme123in", + SymKeyOnly: false, + KeyPath: "", + } + enc, err := New(conf) + if err != nil { + t.Fatalf("cannot create new encoder: %v", err) + } + + packed, err := enc.Encode([]byte(input), []byte("meta"), []byte("META 22222!!!!")) + if err != nil { + t.Fatalf("cannot encode with new encoder: %v", err) + } + unpacked, err := enc.Decode(packed) + if err != nil { + t.Fatalf("cannot decode with new encoder: %v", err) + } + if string(unpacked) != input { + t.Fatalf("unpacked != input: %v vs %v", string(unpacked), input) + } + + packed, err = enc.Encode([]byte(input)) + if err != nil { + t.Fatalf("cannot encode with new encoder: %v", err) + } + unpacked, err = enc.Decode(packed) + if err != nil { + t.Fatalf("cannot decode with new encoder: %v", err) + } + if string(unpacked) != input { + t.Fatalf("unpacked != input: %v vs %v", string(unpacked), input) + } + + t.Logf("prepacked: %q, packed len: %v, unpacked: %q", input, len(packed), string(unpacked)) +} + +var metaname = []byte("META NAME") +var metavalue = []byte("META VALUE") + +func Test_Encoder_BOP_Combos(t *testing.T) { + enc := encryptor.NewEncryptor("", "") + sym := encryptor.NewEncryptor("", "") + sym.SetSymmetric(sym.GetPublic()) + + encAndDec(t, makeSomeEnc(), "default") + encAndDec(t, makeSomeEnc( + &Salter{}, + &asymCryptor{Encryptor: enc}, + &AES{Key: []byte(sym.GetPublic())}, + &Zipper{}, + ), "salt_asym_sym_zip") + encAndDec(t, makeSomeEnc( + &Zipper{}, + &Salter{}, + &asymCryptor{Encryptor: enc}, + &AES{Key: []byte(sym.GetPublic())}, + &Zipper{}, + ), "zip_salt_asym_sym_zip") + encAndDec(t, makeSomeEnc( + &snapper{}, + &Salter{}, + &asymCryptor{Encryptor: enc}, + &AES{Key: []byte(sym.GetPublic())}, + &snapper{}, + ), "snap_salt_asym_sym_snap") + encAndDec(t, makeSomeEnc( + &Salter{}, + &asymCryptor{Encryptor: enc}, + &AES{Key: []byte(sym.GetPublic())}, + &Zipper{}, + ), "salt_asym_sym_zip") + encAndDec(t, makeSomeEnc( + &Zipper{}, + &Salter{}, + &asymCryptor{Encryptor: enc}, + &AES{Key: []byte(sym.GetPublic())}, + ), "zip_salt_asym_sym") + encAndDec(t, makeSomeEnc( + &Zipper{}, + &Salter{}, + &AES{Key: []byte(sym.GetPublic())}, + &asymCryptor{Encryptor: enc}, + &Zipper{}, + ), "zip_salt_sym_asym_zip") + encAndDec(t, makeSomeEnc( + &Zipper{}, + &Salter{n: 50}, + &AES{Key: []byte(sym.GetPublic())}, + &asymCryptor{Encryptor: enc}, + &Zipper{}, + ), "zip_salt50_sym_asym_zip") + encAndDec(t, makeSomeEnc( + &Zipper{}, + &Salter{n: 10}, + &AES{Key: []byte(sym.GetPublic())}, + &asymCryptor{Encryptor: enc}, + &Zipper{}, + ), "zip_salt10_sym_asym_zip") + syms := []encryptor.Encryptor{} + for i := 0; i < 10; i++ { + e := encryptor.NewEncryptor("", "") + e.SetSymmetric(e.GetPublic()) + syms = append(syms, e) + } + encAndDec(t, makeSomeEnc( + &Zipper{}, + &Salter{n: 10}, + &AES{Key: []byte(syms[0].GetPublic())}, + &AES{Key: []byte(syms[1].GetPublic())}, + &AES{Key: []byte(syms[2].GetPublic())}, + &AES{Key: []byte(syms[3].GetPublic())}, + &AES{Key: []byte(syms[4].GetPublic())}, + &AES{Key: []byte(syms[5].GetPublic())}, + &Zipper{}, + ), "zip_salt10_sym6_zip") + encAndDec(t, makeSomeEnc( + &Salter{n: 10}, + &Zipper{}, + &AES{Key: []byte(sym.GetPublic())}, + &Zipper{}, + ), "salt10_zip_sym_zip") + encAndDec(t, makeSomeEnc( + &Salter{n: 10}, + &LZ4{}, + &AES{Key: []byte(sym.GetPublic())}, + &LZ4{}, + ), "salt10_lz4_sym_lz4") + encAndDec(t, makeSomeEnc( + &LZ4{}, + &Salter{n: 10}, + &AES{Key: []byte(sym.GetPublic())}, + &LZ4{}, + ), "salt10_lz4_sym_lz4") + encAndDec(t, makeSomeEnc( + &LZ4{}, + ), "lz4") + encAndDec(t, makeSomeEnc( + &Zipper{}, + ), "zip") +} + +func encAndDec(t *testing.T, enc *Encoder, name string) { + input := readSomeInput() + packed, err := enc.Encode(input, metaname, metavalue) + if err != nil { + t.Fatal(err) + } + unpacked, err := enc.Decode(packed) + if err != nil { + t.Fatal(err) + } + if string(unpacked) != string(input) { + t.Fatalf("Unpacked != input: len %v vs %v", len(unpacked), len(input)) + } + t.Logf("%30q: %4d -> %4d when packed (%d%%)", name, len(input), len(packed), 100.0*len(packed)/len(input)) +} + +func makeSomeEnc(bops ...BOP) *Encoder { + conf := &Config{ + SymKey: "letme123in", + SymKeyOnly: false, + KeyPath: "", + } + e, err := New(conf, bops...) + if err != nil { + panic(err) + } + return e +} + +func readSomeInput() []byte { + b, err := ioutil.ReadFile("./encoder.go") + if err != nil { + panic(err) + } + return b +} diff --git a/pipe/pipe.go b/pipe/pipe.go new file mode 100644 index 0000000..048d636 --- /dev/null +++ b/pipe/pipe.go @@ -0,0 +1,195 @@ +package pipe + +import "io" + +type PipeReader struct { + last []byte + semaphore chan struct{} + hasWrite chan struct{} + closed bool + buffer []byte + lastWrite int + capacity int64 + r io.ReadCloser + w io.WriteCloser +} + +func New() *PipeReader { + r, w := io.Pipe() + return &PipeReader{ + semaphore: make(chan struct{}, 1), + hasWrite: make(chan struct{}), + buffer: make([]byte, 64000), + capacity: 64000, + last: nil, + r: r, + w: w, + } +} + +func (pr *PipeReader) Read(b []byte) (int, error) { + return pr.r.Read(b) +} + +func (pr *PipeReader) Write(b []byte) (int, error) { + chunks := int(int64(len(b)) / pr.capacity) + if int64(len(b))%pr.capacity > 0 { + chunks += 1 + } + ttl := 0 + for i := 0; i < chunks-1; i++ { + n, err := pr.w.Write(b[ttl:pr.capacity]) + if err != nil { + return ttl, err + } + ttl += n + } + n, err := pr.w.Write(b[ttl:]) + return ttl + n, err +} + +func (pr *PipeReader) Close() error { + return pr.w.Close() +} + +func (pr *PipeReader) SetCap(n int64) { + pr.capacity = n +} + +/* + +type PipeReader struct { + last []byte + semaphore chan struct{} + hasWrite chan struct{} + closed bool + buffer []byte + lastWrite int + capacity int +} + +func New() *PipeReader { + return &PipeReader{ + semaphore: make(chan struct{}, 1), + hasWrite: make(chan struct{}), + buffer: make([]byte, 64000), + capacity: 64000, + last: nil, + } +} + +func (pr *PipeReader) SetCap(n int) { + pr.capacity = n + pr.buffer = make([]byte, n) +} + +func (pr *PipeReader) String() string { + return fmt.Sprintf("buf=%v, cap=%v, last=%v", len(pr.buffer), pr.capacity, len(pr.last)) +} + +func (pr *PipeReader) lock() { + pr.semaphore <- struct{}{} +} + +func (pr *PipeReader) unlock() { + for _ = range pr.semaphore { + return + } +} + +func (pr *PipeReader) blockUntilWrite() { + for _ = range pr.hasWrite { + return + } +} + +func (pr *PipeReader) unblockOnWrite() { + pr.hasWrite <- struct{}{} +} + +func (pr *PipeReader) readCurrentBuffer() []byte { + pr.blockUntilWrite() + pr.lock() + defer pr.unlock() + ret := pr.buffer[:pr.lastWrite] + pr.lastWrite = 0 + return ret +} + +func (pr *PipeReader) writeBuffer(b []byte) { + pr.lock() + defer pr.unblockOnWrite() + defer pr.unlock() + copy(pr.buffer[:len(b)], b) + pr.lastWrite = len(b) +} + +func (pr *PipeReader) Read(b []byte) (int, error) { + logger.Log("READ", len(b), pr.isClosed()) + defer logger.Log("/READ") + ttl := 0 + m, n, hasMore := xfer(b, pr.last) + pr.last = m + ttl = n + if hasMore { + return ttl, nil + } + for { + m, n, hasMore := xfer(b[ttl:], pr.readCurrentBuffer()) + logger.Log("XFER", m, n, hasMore, ttl, pr.isClosed()) + pr.last = m + ttl += n + if hasMore { + return ttl, nil + } + if n == 0 && pr.isClosed() { + return ttl, io.EOF + } + } + return ttl, io.EOF +} + +func (pr *PipeReader) Write(b []byte) (int, error) { + logger.Log("WRIT", len(b), pr.isClosed()) + defer logger.Log("/WRIT") + chunklen := pr.capacity + wrote := 0 + for wrote+chunklen < len(b) { + pr.writeBuffer(b[wrote : wrote+chunklen]) + wrote += chunklen + } + pr.writeBuffer(b[wrote:]) + return wrote + len(b[wrote:]), nil +} + +func (pr *PipeReader) isClosed() bool { + pr.lock() + defer pr.unlock() + return pr.closed +} + +func (pr *PipeReader) Close() error { + pr.lock() + pr.closed = true + close(pr.hasWrite) + pr.unlock() + return nil +} + +func copyByteSlice(b []byte) []byte { + return append([]byte{}, b...) +} + +func xfer(dst, src []byte) ([]byte, int, bool) { + min := len(dst) + if len(src) < min { + min = len(src) + } + copy(dst[:min], src[:min]) + src = src[min:] + if len(src) > 0 { + return src, min, true + } + return src, min, false +} +*/ diff --git a/pipe/pipe_test.go b/pipe/pipe_test.go new file mode 100644 index 0000000..611246f --- /dev/null +++ b/pipe/pipe_test.go @@ -0,0 +1,95 @@ +package pipe + +import ( + "crypto/rand" + "io/ioutil" + "strings" + "testing" +) + +func Test_PipeReader_Bench(t *testing.T) { + input := []byte(strings.Repeat("hello world", 1000)) + pr := New() + pr.SetCap(1000) + writing := func() { + defer pr.Close() + if n, err := pr.Write(input); err != nil { + t.Errorf("failed to write: %v", err) + } else if n != len(input) { + t.Errorf("failed to write: wrote %v, expected %v", n, len(input)) + } + } + reading := func() { + if bytes, err := ioutil.ReadAll(pr); err != nil { + t.Errorf("failed to read: %v", err) + } else if len(bytes) != len(input) { + t.Errorf("failed to read: read %v, expected %v", len(bytes), len(input)) + } + } + go writing() + reading() +} + +func Test_PipeReader(t *testing.T) { + cases := []struct { + streamLength int + bufferCap int + }{ + { + streamLength: 10000000, + bufferCap: 50, + }, + { + streamLength: 1000000, + bufferCap: 5000, + }, + { + streamLength: 100, + bufferCap: 50, + }, + { + streamLength: 50, + bufferCap: 100, + }, + { + streamLength: 10, + bufferCap: 5, + }, + { + streamLength: 5, + bufferCap: 10, + }, + { + streamLength: 1, + bufferCap: 10, + }, + { + streamLength: 0, + bufferCap: 10, + }, + } + + for _, c := range cases { + pr := New() + pr.SetCap(int64(c.bufferCap)) + input := make([]byte, c.streamLength) + if _, err := rand.Read(input); err != nil { + t.Fatalf("cannot read random bytes: %v", err) + } + go func() { + defer pr.Close() + n, err := pr.Write(input) + if n != c.streamLength || err != nil { + t.Errorf("ERR: failed to write %v bytes to %v cap: %v", c.streamLength, c.bufferCap, err) + } + t.Logf("wrote %v bytes to pipereader", n) + }() + output, err := ioutil.ReadAll(pr) + if err != nil { + t.Errorf("ERR: failed to read %v bytes to %v cap: %v", c.streamLength, c.bufferCap, err) + } else if string(output) != string(input) { + t.Errorf("ERR: sl=%v, bc=%v, read wrong output from pipe: \nwanted %v, \n got %v", c.streamLength, c.bufferCap, len(input), len(output)) + } + t.Logf("read %v bytes from pipereader", len(output)) + } +}