You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

unsnap.go 14KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519
  1. package unsnap
  2. import (
  3. "bytes"
  4. "encoding/binary"
  5. "fmt"
  6. "io"
  7. "io/ioutil"
  8. "os"
  9. "strings"
  10. "hash/crc32"
  11. snappy "github.com/golang/snappy"
  12. // The C library can be used, but this makes the binary dependent
  13. // lots of extraneous c-libraries; it is no longer stand-alone. Yuck.
  14. //
  15. // Therefore we comment out the "dgryski/go-csnappy" path and use the
  16. // "github.com/golang/snappy/snappy" above instead. If you are
  17. // performance limited and can deal with distributing more libraries,
  18. // then this is easy to swap.
  19. //
  20. // If you swap, note that some of the tests won't pass
  21. // because snappy-go produces slightly different (but still
  22. // conformant) encodings on some data. Here are bindings
  23. // to the C-snappy:
  24. // snappy "github.com/dgryski/go-csnappy"
  25. )
  26. // SnappyFile: create a drop-in-replacement/wrapper for an *os.File that handles doing the unsnappification online as more is read from it
  27. type SnappyFile struct {
  28. Fname string
  29. Reader io.Reader
  30. Writer io.Writer
  31. // allow clients to substitute us for an os.File and just switch
  32. // off compression if they don't want it.
  33. SnappyEncodeDecodeOff bool // if true, we bypass straight to Filep
  34. EncBuf FixedSizeRingBuf // holds any extra that isn't yet returned, encoded
  35. DecBuf FixedSizeRingBuf // holds any extra that isn't yet returned, decoded
  36. // for writing to stream-framed snappy
  37. HeaderChunkWritten bool
  38. // Sanity check: we can only read, or only write, to one SnappyFile.
  39. // EncBuf and DecBuf are used differently in each mode. Verify
  40. // that we are consistent with this flag.
  41. Writing bool
  42. }
  43. var total int
  44. // for debugging, show state of buffers
  45. func (f *SnappyFile) Dump() {
  46. fmt.Printf("EncBuf has length %d and contents:\n%s\n", len(f.EncBuf.Bytes()), string(f.EncBuf.Bytes()))
  47. fmt.Printf("DecBuf has length %d and contents:\n%s\n", len(f.DecBuf.Bytes()), string(f.DecBuf.Bytes()))
  48. }
  49. func (f *SnappyFile) Read(p []byte) (n int, err error) {
  50. if f.SnappyEncodeDecodeOff {
  51. return f.Reader.Read(p)
  52. }
  53. if f.Writing {
  54. panic("Reading on a write-only SnappyFile")
  55. }
  56. // before we unencrypt more, try to drain the DecBuf first
  57. n, _ = f.DecBuf.Read(p)
  58. if n > 0 {
  59. total += n
  60. return n, nil
  61. }
  62. //nEncRead, nDecAdded, err := UnsnapOneFrame(f.Filep, &f.EncBuf, &f.DecBuf, f.Fname)
  63. _, _, err = UnsnapOneFrame(f.Reader, &f.EncBuf, &f.DecBuf, f.Fname)
  64. if err != nil && err != io.EOF {
  65. panic(err)
  66. }
  67. n, _ = f.DecBuf.Read(p)
  68. if n > 0 {
  69. total += n
  70. return n, nil
  71. }
  72. if f.DecBuf.Readable == 0 {
  73. if f.DecBuf.Readable == 0 && f.EncBuf.Readable == 0 {
  74. // only now (when EncBuf is empty) can we give io.EOF.
  75. // Any earlier, and we leave stuff un-decoded!
  76. return 0, io.EOF
  77. }
  78. }
  79. return 0, nil
  80. }
  81. func Open(name string) (file *SnappyFile, err error) {
  82. fp, err := os.Open(name)
  83. if err != nil {
  84. return nil, err
  85. }
  86. // encoding in snappy can apparently go beyond the original size, so
  87. // we make our buffers big enough, 2*max snappy chunk => 2 * CHUNK_MAX(65536)
  88. snap := NewReader(fp)
  89. snap.Fname = name
  90. return snap, nil
  91. }
  92. func NewReader(r io.Reader) *SnappyFile {
  93. return &SnappyFile{
  94. Reader: r,
  95. EncBuf: *NewFixedSizeRingBuf(CHUNK_MAX * 2), // buffer of snappy encoded bytes
  96. DecBuf: *NewFixedSizeRingBuf(CHUNK_MAX * 2), // buffer of snapppy decoded bytes
  97. Writing: false,
  98. }
  99. }
  100. func NewWriter(w io.Writer) *SnappyFile {
  101. return &SnappyFile{
  102. Writer: w,
  103. EncBuf: *NewFixedSizeRingBuf(65536), // on writing: temp for testing compression
  104. DecBuf: *NewFixedSizeRingBuf(65536 * 2), // on writing: final buffer of snappy framed and encoded bytes
  105. Writing: true,
  106. }
  107. }
  108. func Create(name string) (file *SnappyFile, err error) {
  109. fp, err := os.Create(name)
  110. if err != nil {
  111. return nil, err
  112. }
  113. snap := NewWriter(fp)
  114. snap.Fname = name
  115. return snap, nil
  116. }
  117. func (f *SnappyFile) Close() error {
  118. if f.Writing {
  119. wc, ok := f.Writer.(io.WriteCloser)
  120. if ok {
  121. return wc.Close()
  122. }
  123. return nil
  124. }
  125. rc, ok := f.Reader.(io.ReadCloser)
  126. if ok {
  127. return rc.Close()
  128. }
  129. return nil
  130. }
  131. func (f *SnappyFile) Sync() error {
  132. file, ok := f.Writer.(*os.File)
  133. if ok {
  134. return file.Sync()
  135. }
  136. return nil
  137. }
  138. // for an increment of a frame at a time:
  139. // read from r into encBuf (encBuf is still encoded, thus the name), and write unsnappified frames into outDecodedBuf
  140. // the returned n: number of bytes read from the encrypted encBuf
  141. func UnsnapOneFrame(r io.Reader, encBuf *FixedSizeRingBuf, outDecodedBuf *FixedSizeRingBuf, fname string) (nEnc int64, nDec int64, err error) {
  142. // b, err := ioutil.ReadAll(r)
  143. // if err != nil {
  144. // panic(err)
  145. // }
  146. nEnc = 0
  147. nDec = 0
  148. // read up to 65536 bytes from r into encBuf, at least a snappy frame
  149. nread, err := io.CopyN(encBuf, r, 65536) // returns nwrotebytes, err
  150. nEnc += nread
  151. if err != nil {
  152. if err == io.EOF {
  153. if nread == 0 {
  154. if encBuf.Readable == 0 {
  155. return nEnc, nDec, io.EOF
  156. }
  157. // else we have bytes in encBuf, so decode them!
  158. err = nil
  159. } else {
  160. // continue below, processing the nread bytes
  161. err = nil
  162. }
  163. } else {
  164. // may be an odd already closed... don't panic on that
  165. if strings.Contains(err.Error(), "file already closed") {
  166. err = nil
  167. } else {
  168. panic(err)
  169. }
  170. }
  171. }
  172. // flag for printing chunk size alignment messages
  173. verbose := false
  174. const snappyStreamHeaderSz = 10
  175. const headerSz = 4
  176. const crc32Sz = 4
  177. // the magic 18 bytes accounts for the snappy streaming header and the first chunks size and checksum
  178. // http://code.google.com/p/snappy/source/browse/trunk/framing_format.txt
  179. chunk := (*encBuf).Bytes()
  180. // however we exit, advance as
  181. // defer func() { (*encBuf).Next(N) }()
  182. // 65536 is the max size of a snappy framed chunk. See
  183. // http://code.google.com/p/snappy/source/browse/trunk/framing_format.txt:91
  184. // buf := make([]byte, 65536)
  185. // fmt.Printf("read from file, b is len:%d with value: %#v\n", len(b), b)
  186. // fmt.Printf("read from file, bcut is len:%d with value: %#v\n", len(bcut), bcut)
  187. //fmt.Printf("raw bytes of chunksz are: %v\n", b[11:14])
  188. fourbytes := make([]byte, 4)
  189. chunkCount := 0
  190. for nDec < 65536 {
  191. if len(chunk) == 0 {
  192. break
  193. }
  194. chunkCount++
  195. fourbytes[3] = 0
  196. copy(fourbytes, chunk[1:4])
  197. chunksz := binary.LittleEndian.Uint32(fourbytes)
  198. chunk_type := chunk[0]
  199. switch true {
  200. case chunk_type == 0xff:
  201. { // stream identifier
  202. streamHeader := chunk[:snappyStreamHeaderSz]
  203. if 0 != bytes.Compare(streamHeader, []byte{0xff, 0x06, 0x00, 0x00, 0x73, 0x4e, 0x61, 0x50, 0x70, 0x59}) {
  204. panic("file had chunk starting with 0xff but then no magic snappy streaming protocol bytes, aborting.")
  205. } else {
  206. //fmt.Printf("got streaming snappy magic header just fine.\n")
  207. }
  208. chunk = chunk[snappyStreamHeaderSz:]
  209. (*encBuf).Advance(snappyStreamHeaderSz)
  210. nEnc += snappyStreamHeaderSz
  211. continue
  212. }
  213. case chunk_type == 0x00:
  214. { // compressed data
  215. if verbose {
  216. fmt.Fprintf(os.Stderr, "chunksz is %d while total bytes avail are: %d\n", int(chunksz), len(chunk)-4)
  217. }
  218. crc := binary.LittleEndian.Uint32(chunk[headerSz:(headerSz + crc32Sz)])
  219. section := chunk[(headerSz + crc32Sz):(headerSz + chunksz)]
  220. dec, ok := snappy.Decode(nil, section)
  221. if ok != nil {
  222. // we've probably truncated a snappy frame at this point
  223. // ok=snappy: corrupt input
  224. // len(dec) == 0
  225. //
  226. panic(fmt.Sprintf("could not decode snappy stream: '%s' and len dec=%d and ok=%v\n", fname, len(dec), ok))
  227. // get back to caller with what we've got so far
  228. return nEnc, nDec, nil
  229. }
  230. // fmt.Printf("ok, b is %#v , %#v\n", ok, dec)
  231. // spit out decoded text
  232. // n, err := w.Write(dec)
  233. //fmt.Printf("len(dec) = %d, outDecodedBuf.Readable=%d\n", len(dec), outDecodedBuf.Readable)
  234. bnb := bytes.NewBuffer(dec)
  235. n, err := io.Copy(outDecodedBuf, bnb)
  236. if err != nil {
  237. //fmt.Printf("got n=%d, err= %s ; when trying to io.Copy(outDecodedBuf: N=%d, Readable=%d)\n", n, err, outDecodedBuf.N, outDecodedBuf.Readable)
  238. panic(err)
  239. }
  240. if n != int64(len(dec)) {
  241. panic("could not write all bytes to outDecodedBuf")
  242. }
  243. nDec += n
  244. // verify the crc32 rotated checksum
  245. m32 := masked_crc32c(dec)
  246. if m32 != crc {
  247. panic(fmt.Sprintf("crc32 masked failiure. expected: %v but got: %v", crc, m32))
  248. } else {
  249. //fmt.Printf("\nchecksums match: %v == %v\n", crc, m32)
  250. }
  251. // move to next header
  252. inc := (headerSz + int(chunksz))
  253. chunk = chunk[inc:]
  254. (*encBuf).Advance(inc)
  255. nEnc += int64(inc)
  256. continue
  257. }
  258. case chunk_type == 0x01:
  259. { // uncompressed data
  260. //n, err := w.Write(chunk[(headerSz+crc32Sz):(headerSz + int(chunksz))])
  261. n, err := io.Copy(outDecodedBuf, bytes.NewBuffer(chunk[(headerSz+crc32Sz):(headerSz+int(chunksz))]))
  262. if verbose {
  263. //fmt.Printf("debug: n=%d err=%v chunksz=%d outDecodedBuf='%v'\n", n, err, chunksz, outDecodedBuf)
  264. }
  265. if err != nil {
  266. panic(err)
  267. }
  268. if n != int64(chunksz-crc32Sz) {
  269. panic("could not write all bytes to stdout")
  270. }
  271. nDec += n
  272. inc := (headerSz + int(chunksz))
  273. chunk = chunk[inc:]
  274. (*encBuf).Advance(inc)
  275. nEnc += int64(inc)
  276. continue
  277. }
  278. case chunk_type == 0xfe:
  279. fallthrough // padding, just skip it
  280. case chunk_type >= 0x80 && chunk_type <= 0xfd:
  281. { // Reserved skippable chunks
  282. //fmt.Printf("\nin reserved skippable chunks, at nEnc=%v\n", nEnc)
  283. inc := (headerSz + int(chunksz))
  284. chunk = chunk[inc:]
  285. nEnc += int64(inc)
  286. (*encBuf).Advance(inc)
  287. continue
  288. }
  289. default:
  290. panic(fmt.Sprintf("unrecognized/unsupported chunk type %#v", chunk_type))
  291. }
  292. } // end for{}
  293. return nEnc, nDec, err
  294. //return int64(N), nil
  295. }
  296. // for whole file at once:
  297. //
  298. // receive on stdin a stream of bytes in the snappy-streaming framed
  299. // format, defined here: http://code.google.com/p/snappy/source/browse/trunk/framing_format.txt
  300. // Grab each frame, run it through the snappy decoder, and spit out
  301. // each frame all joined back-to-back on stdout.
  302. //
  303. func Unsnappy(r io.Reader, w io.Writer) (err error) {
  304. b, err := ioutil.ReadAll(r)
  305. if err != nil {
  306. panic(err)
  307. }
  308. // flag for printing chunk size alignment messages
  309. verbose := false
  310. const snappyStreamHeaderSz = 10
  311. const headerSz = 4
  312. const crc32Sz = 4
  313. // the magic 18 bytes accounts for the snappy streaming header and the first chunks size and checksum
  314. // http://code.google.com/p/snappy/source/browse/trunk/framing_format.txt
  315. chunk := b[:]
  316. // 65536 is the max size of a snappy framed chunk. See
  317. // http://code.google.com/p/snappy/source/browse/trunk/framing_format.txt:91
  318. //buf := make([]byte, 65536)
  319. // fmt.Printf("read from file, b is len:%d with value: %#v\n", len(b), b)
  320. // fmt.Printf("read from file, bcut is len:%d with value: %#v\n", len(bcut), bcut)
  321. //fmt.Printf("raw bytes of chunksz are: %v\n", b[11:14])
  322. fourbytes := make([]byte, 4)
  323. chunkCount := 0
  324. for {
  325. if len(chunk) == 0 {
  326. break
  327. }
  328. chunkCount++
  329. fourbytes[3] = 0
  330. copy(fourbytes, chunk[1:4])
  331. chunksz := binary.LittleEndian.Uint32(fourbytes)
  332. chunk_type := chunk[0]
  333. switch true {
  334. case chunk_type == 0xff:
  335. { // stream identifier
  336. streamHeader := chunk[:snappyStreamHeaderSz]
  337. if 0 != bytes.Compare(streamHeader, []byte{0xff, 0x06, 0x00, 0x00, 0x73, 0x4e, 0x61, 0x50, 0x70, 0x59}) {
  338. panic("file had chunk starting with 0xff but then no magic snappy streaming protocol bytes, aborting.")
  339. } else {
  340. //fmt.Printf("got streaming snappy magic header just fine.\n")
  341. }
  342. chunk = chunk[snappyStreamHeaderSz:]
  343. continue
  344. }
  345. case chunk_type == 0x00:
  346. { // compressed data
  347. if verbose {
  348. fmt.Fprintf(os.Stderr, "chunksz is %d while total bytes avail are: %d\n", int(chunksz), len(chunk)-4)
  349. }
  350. //crc := binary.LittleEndian.Uint32(chunk[headerSz:(headerSz + crc32Sz)])
  351. section := chunk[(headerSz + crc32Sz):(headerSz + chunksz)]
  352. dec, ok := snappy.Decode(nil, section)
  353. if ok != nil {
  354. panic("could not decode snappy stream")
  355. }
  356. // fmt.Printf("ok, b is %#v , %#v\n", ok, dec)
  357. // spit out decoded text
  358. n, err := w.Write(dec)
  359. if err != nil {
  360. panic(err)
  361. }
  362. if n != len(dec) {
  363. panic("could not write all bytes to stdout")
  364. }
  365. // TODO: verify the crc32 rotated checksum?
  366. // move to next header
  367. chunk = chunk[(headerSz + int(chunksz)):]
  368. continue
  369. }
  370. case chunk_type == 0x01:
  371. { // uncompressed data
  372. //crc := binary.LittleEndian.Uint32(chunk[headerSz:(headerSz + crc32Sz)])
  373. section := chunk[(headerSz + crc32Sz):(headerSz + chunksz)]
  374. n, err := w.Write(section)
  375. if err != nil {
  376. panic(err)
  377. }
  378. if n != int(chunksz-crc32Sz) {
  379. panic("could not write all bytes to stdout")
  380. }
  381. chunk = chunk[(headerSz + int(chunksz)):]
  382. continue
  383. }
  384. case chunk_type == 0xfe:
  385. fallthrough // padding, just skip it
  386. case chunk_type >= 0x80 && chunk_type <= 0xfd:
  387. { // Reserved skippable chunks
  388. chunk = chunk[(headerSz + int(chunksz)):]
  389. continue
  390. }
  391. default:
  392. panic(fmt.Sprintf("unrecognized/unsupported chunk type %#v", chunk_type))
  393. }
  394. } // end for{}
  395. return nil
  396. }
  397. // 0xff 0x06 0x00 0x00 sNaPpY
  398. var SnappyStreamHeaderMagic = []byte{0xff, 0x06, 0x00, 0x00, 0x73, 0x4e, 0x61, 0x50, 0x70, 0x59}
  399. const CHUNK_MAX = 65536
  400. const _STREAM_TO_STREAM_BLOCK_SIZE = CHUNK_MAX
  401. const _STREAM_IDENTIFIER = `sNaPpY`
  402. const _COMPRESSED_CHUNK = 0x00
  403. const _UNCOMPRESSED_CHUNK = 0x01
  404. const _IDENTIFIER_CHUNK = 0xff
  405. const _RESERVED_UNSKIPPABLE0 = 0x02 // chunk ranges are [inclusive, exclusive)
  406. const _RESERVED_UNSKIPPABLE1 = 0x80
  407. const _RESERVED_SKIPPABLE0 = 0x80
  408. const _RESERVED_SKIPPABLE1 = 0xff
  409. // the minimum percent of bytes compression must save to be enabled in automatic
  410. // mode
  411. const _COMPRESSION_THRESHOLD = .125
  412. var crctab *crc32.Table
  413. func init() {
  414. crctab = crc32.MakeTable(crc32.Castagnoli) // this is correct table, matches the crc32c.c code used by python
  415. }
  416. func masked_crc32c(data []byte) uint32 {
  417. // see the framing format specification, http://code.google.com/p/snappy/source/browse/trunk/framing_format.txt
  418. var crc uint32 = crc32.Checksum(data, crctab)
  419. return (uint32((crc>>15)|(crc<<17)) + 0xa282ead8)
  420. }
  421. func ReadSnappyStreamCompressedFile(filename string) ([]byte, error) {
  422. snappyFile, err := Open(filename)
  423. if err != nil {
  424. return []byte{}, err
  425. }
  426. var bb bytes.Buffer
  427. _, err = bb.ReadFrom(snappyFile)
  428. if err == io.EOF {
  429. err = nil
  430. }
  431. if err != nil {
  432. panic(err)
  433. }
  434. return bb.Bytes(), err
  435. }