summaryrefslogtreecommitdiffstats
path: root/compat/zlib/examples/gun.c
diff options
context:
space:
mode:
Diffstat (limited to 'compat/zlib/examples/gun.c')
-rw-r--r--compat/zlib/examples/gun.c702
1 files changed, 702 insertions, 0 deletions
diff --git a/compat/zlib/examples/gun.c b/compat/zlib/examples/gun.c
new file mode 100644
index 0000000..89e484f
--- /dev/null
+++ b/compat/zlib/examples/gun.c
@@ -0,0 +1,702 @@
+/* gun.c -- simple gunzip to give an example of the use of inflateBack()
+ * Copyright (C) 2003, 2005, 2008, 2010, 2012 Mark Adler
+ * For conditions of distribution and use, see copyright notice in zlib.h
+ Version 1.7 12 August 2012 Mark Adler */
+
+/* Version history:
+ 1.0 16 Feb 2003 First version for testing of inflateBack()
+ 1.1 21 Feb 2005 Decompress concatenated gzip streams
+ Remove use of "this" variable (C++ keyword)
+ Fix return value for in()
+ Improve allocation failure checking
+ Add typecasting for void * structures
+ Add -h option for command version and usage
+ Add a bunch of comments
+ 1.2 20 Mar 2005 Add Unix compress (LZW) decompression
+ Copy file attributes from input file to output file
+ 1.3 12 Jun 2005 Add casts for error messages [Oberhumer]
+ 1.4 8 Dec 2006 LZW decompression speed improvements
+ 1.5 9 Feb 2008 Avoid warning in latest version of gcc
+ 1.6 17 Jan 2010 Avoid signed/unsigned comparison warnings
+ 1.7 12 Aug 2012 Update for z_const usage in zlib 1.2.8
+ */
+
+/*
+ gun [ -t ] [ name ... ]
+
+ decompresses the data in the named gzip files. If no arguments are given,
+ gun will decompress from stdin to stdout. The names must end in .gz, -gz,
+ .z, -z, _z, or .Z. The uncompressed data will be written to a file name
+ with the suffix stripped. On success, the original file is deleted. On
+ failure, the output file is deleted. For most failures, the command will
+ continue to process the remaining names on the command line. A memory
+ allocation failure will abort the command. If -t is specified, then the
+ listed files or stdin will be tested as gzip files for integrity (without
+ checking for a proper suffix), no output will be written, and no files
+ will be deleted.
+
+ Like gzip, gun allows concatenated gzip streams and will decompress them,
+ writing all of the uncompressed data to the output. Unlike gzip, gun allows
+ an empty file on input, and will produce no error writing an empty output
+ file.
+
+ gun will also decompress files made by Unix compress, which uses LZW
+ compression. These files are automatically detected by virtue of their
+ magic header bytes. Since the end of Unix compress stream is marked by the
+ end-of-file, they cannot be concantenated. If a Unix compress stream is
+ encountered in an input file, it is the last stream in that file.
+
+ Like gunzip and uncompress, the file attributes of the orignal compressed
+ file are maintained in the final uncompressed file, to the extent that the
+ user permissions allow it.
+
+ On my Mac OS X PowerPC G4, gun is almost twice as fast as gunzip (version
+ 1.2.4) is on the same file, when gun is linked with zlib 1.2.2. Also the
+ LZW decompression provided by gun is about twice as fast as the standard
+ Unix uncompress command.
+ */
+
+/* external functions and related types and constants */
+#include <stdio.h> /* fprintf() */
+#include <stdlib.h> /* malloc(), free() */
+#include <string.h> /* strerror(), strcmp(), strlen(), memcpy() */
+#include <errno.h> /* errno */
+#include <fcntl.h> /* open() */
+#include <unistd.h> /* read(), write(), close(), chown(), unlink() */
+#include <sys/types.h>
+#include <sys/stat.h> /* stat(), chmod() */
+#include <utime.h> /* utime() */
+#include "zlib.h" /* inflateBackInit(), inflateBack(), */
+ /* inflateBackEnd(), crc32() */
+
+/* function declaration */
+#define local static
+
+/* buffer constants */
+#define SIZE 32768U /* input and output buffer sizes */
+#define PIECE 16384 /* limits i/o chunks for 16-bit int case */
+
+/* structure for infback() to pass to input function in() -- it maintains the
+ input file and a buffer of size SIZE */
+struct ind {
+ int infile;
+ unsigned char *inbuf;
+};
+
+/* Load input buffer, assumed to be empty, and return bytes loaded and a
+ pointer to them. read() is called until the buffer is full, or until it
+ returns end-of-file or error. Return 0 on error. */
+local unsigned in(void *in_desc, z_const unsigned char **buf)
+{
+ int ret;
+ unsigned len;
+ unsigned char *next;
+ struct ind *me = (struct ind *)in_desc;
+
+ next = me->inbuf;
+ *buf = next;
+ len = 0;
+ do {
+ ret = PIECE;
+ if ((unsigned)ret > SIZE - len)
+ ret = (int)(SIZE - len);
+ ret = (int)read(me->infile, next, ret);
+ if (ret == -1) {
+ len = 0;
+ break;
+ }
+ next += ret;
+ len += ret;
+ } while (ret != 0 && len < SIZE);
+ return len;
+}
+
+/* structure for infback() to pass to output function out() -- it maintains the
+ output file, a running CRC-32 check on the output and the total number of
+ bytes output, both for checking against the gzip trailer. (The length in
+ the gzip trailer is stored modulo 2^32, so it's ok if a long is 32 bits and
+ the output is greater than 4 GB.) */
+struct outd {
+ int outfile;
+ int check; /* true if checking crc and total */
+ unsigned long crc;
+ unsigned long total;
+};
+
+/* Write output buffer and update the CRC-32 and total bytes written. write()
+ is called until all of the output is written or an error is encountered.
+ On success out() returns 0. For a write failure, out() returns 1. If the
+ output file descriptor is -1, then nothing is written.
+ */
+local int out(void *out_desc, unsigned char *buf, unsigned len)
+{
+ int ret;
+ struct outd *me = (struct outd *)out_desc;
+
+ if (me->check) {
+ me->crc = crc32(me->crc, buf, len);
+ me->total += len;
+ }
+ if (me->outfile != -1)
+ do {
+ ret = PIECE;
+ if ((unsigned)ret > len)
+ ret = (int)len;
+ ret = (int)write(me->outfile, buf, ret);
+ if (ret == -1)
+ return 1;
+ buf += ret;
+ len -= ret;
+ } while (len != 0);
+ return 0;
+}
+
+/* next input byte macro for use inside lunpipe() and gunpipe() */
+#define NEXT() (have ? 0 : (have = in(indp, &next)), \
+ last = have ? (have--, (int)(*next++)) : -1)
+
+/* memory for gunpipe() and lunpipe() --
+ the first 256 entries of prefix[] and suffix[] are never used, could
+ have offset the index, but it's faster to waste the memory */
+unsigned char inbuf[SIZE]; /* input buffer */
+unsigned char outbuf[SIZE]; /* output buffer */
+unsigned short prefix[65536]; /* index to LZW prefix string */
+unsigned char suffix[65536]; /* one-character LZW suffix */
+unsigned char match[65280 + 2]; /* buffer for reversed match or gzip
+ 32K sliding window */
+
+/* throw out what's left in the current bits byte buffer (this is a vestigial
+ aspect of the compressed data format derived from an implementation that
+ made use of a special VAX machine instruction!) */
+#define FLUSHCODE() \
+ do { \
+ left = 0; \
+ rem = 0; \
+ if (chunk > have) { \
+ chunk -= have; \
+ have = 0; \
+ if (NEXT() == -1) \
+ break; \
+ chunk--; \
+ if (chunk > have) { \
+ chunk = have = 0; \
+ break; \
+ } \
+ } \
+ have -= chunk; \
+ next += chunk; \
+ chunk = 0; \
+ } while (0)
+
+/* Decompress a compress (LZW) file from indp to outfile. The compress magic
+ header (two bytes) has already been read and verified. There are have bytes
+ of buffered input at next. strm is used for passing error information back
+ to gunpipe().
+
+ lunpipe() will return Z_OK on success, Z_BUF_ERROR for an unexpected end of
+ file, read error, or write error (a write error indicated by strm->next_in
+ not equal to Z_NULL), or Z_DATA_ERROR for invalid input.
+ */
+local int lunpipe(unsigned have, z_const unsigned char *next, struct ind *indp,
+ int outfile, z_stream *strm)
+{
+ int last; /* last byte read by NEXT(), or -1 if EOF */
+ unsigned chunk; /* bytes left in current chunk */
+ int left; /* bits left in rem */
+ unsigned rem; /* unused bits from input */
+ int bits; /* current bits per code */
+ unsigned code; /* code, table traversal index */
+ unsigned mask; /* mask for current bits codes */
+ int max; /* maximum bits per code for this stream */
+ unsigned flags; /* compress flags, then block compress flag */
+ unsigned end; /* last valid entry in prefix/suffix tables */
+ unsigned temp; /* current code */
+ unsigned prev; /* previous code */
+ unsigned final; /* last character written for previous code */
+ unsigned stack; /* next position for reversed string */
+ unsigned outcnt; /* bytes in output buffer */
+ struct outd outd; /* output structure */
+ unsigned char *p;
+
+ /* set up output */
+ outd.outfile = outfile;
+ outd.check = 0;
+
+ /* process remainder of compress header -- a flags byte */
+ flags = NEXT();
+ if (last == -1)
+ return Z_BUF_ERROR;
+ if (flags & 0x60) {
+ strm->msg = (char *)"unknown lzw flags set";
+ return Z_DATA_ERROR;
+ }
+ max = flags & 0x1f;
+ if (max < 9 || max > 16) {
+ strm->msg = (char *)"lzw bits out of range";
+ return Z_DATA_ERROR;
+ }
+ if (max == 9) /* 9 doesn't really mean 9 */
+ max = 10;
+ flags &= 0x80; /* true if block compress */
+
+ /* clear table */
+ bits = 9;
+ mask = 0x1ff;
+ end = flags ? 256 : 255;
+
+ /* set up: get first 9-bit code, which is the first decompressed byte, but
+ don't create a table entry until the next code */
+ if (NEXT() == -1) /* no compressed data is ok */
+ return Z_OK;
+ final = prev = (unsigned)last; /* low 8 bits of code */
+ if (NEXT() == -1) /* missing a bit */
+ return Z_BUF_ERROR;
+ if (last & 1) { /* code must be < 256 */
+ strm->msg = (char *)"invalid lzw code";
+ return Z_DATA_ERROR;
+ }
+ rem = (unsigned)last >> 1; /* remaining 7 bits */
+ left = 7;
+ chunk = bits - 2; /* 7 bytes left in this chunk */
+ outbuf[0] = (unsigned char)final; /* write first decompressed byte */
+ outcnt = 1;
+
+ /* decode codes */
+ stack = 0;
+ for (;;) {
+ /* if the table will be full after this, increment the code size */
+ if (end >= mask && bits < max) {
+ FLUSHCODE();
+ bits++;
+ mask <<= 1;
+ mask++;
+ }
+
+ /* get a code of length bits */
+ if (chunk == 0) /* decrement chunk modulo bits */
+ chunk = bits;
+ code = rem; /* low bits of code */
+ if (NEXT() == -1) { /* EOF is end of compressed data */
+ /* write remaining buffered output */
+ if (outcnt && out(&outd, outbuf, outcnt)) {
+ strm->next_in = outbuf; /* signal write error */
+ return Z_BUF_ERROR;
+ }
+ return Z_OK;
+ }
+ code += (unsigned)last << left; /* middle (or high) bits of code */
+ left += 8;
+ chunk--;
+ if (bits > left) { /* need more bits */
+ if (NEXT() == -1) /* can't end in middle of code */
+ return Z_BUF_ERROR;
+ code += (unsigned)last << left; /* high bits of code */
+ left += 8;
+ chunk--;
+ }
+ code &= mask; /* mask to current code length */
+ left -= bits; /* number of unused bits */
+ rem = (unsigned)last >> (8 - left); /* unused bits from last byte */
+
+ /* process clear code (256) */
+ if (code == 256 && flags) {
+ FLUSHCODE();
+ bits = 9; /* initialize bits and mask */
+ mask = 0x1ff;
+ end = 255; /* empty table */
+ continue; /* get next code */
+ }
+
+ /* special code to reuse last match */
+ temp = code; /* save the current code */
+ if (code > end) {
+ /* Be picky on the allowed code here, and make sure that the code
+ we drop through (prev) will be a valid index so that random
+ input does not cause an exception. The code != end + 1 check is
+ empirically derived, and not checked in the original uncompress
+ code. If this ever causes a problem, that check could be safely
+ removed. Leaving this check in greatly improves gun's ability
+ to detect random or corrupted input after a compress header.
+ In any case, the prev > end check must be retained. */
+ if (code != end + 1 || prev > end) {
+ strm->msg = (char *)"invalid lzw code";
+ return Z_DATA_ERROR;
+ }
+ match[stack++] = (unsigned char)final;
+ code = prev;
+ }
+
+ /* walk through linked list to generate output in reverse order */
+ p = match + stack;
+ while (code >= 256) {
+ *p++ = suffix[code];
+ code = prefix[code];
+ }
+ stack = p - match;
+ match[stack++] = (unsigned char)code;
+ final = code;
+
+ /* link new table entry */
+ if (end < mask) {
+ end++;
+ prefix[end] = (unsigned short)prev;
+ suffix[end] = (unsigned char)final;
+ }
+
+ /* set previous code for next iteration */
+ prev = temp;
+
+ /* write output in forward order */
+ while (stack > SIZE - outcnt) {
+ while (outcnt < SIZE)
+ outbuf[outcnt++] = match[--stack];
+ if (out(&outd, outbuf, outcnt)) {
+ strm->next_in = outbuf; /* signal write error */
+ return Z_BUF_ERROR;
+ }
+ outcnt = 0;
+ }
+ p = match + stack;
+ do {
+ outbuf[outcnt++] = *--p;
+ } while (p > match);
+ stack = 0;
+
+ /* loop for next code with final and prev as the last match, rem and
+ left provide the first 0..7 bits of the next code, end is the last
+ valid table entry */
+ }
+}
+
+/* Decompress a gzip file from infile to outfile. strm is assumed to have been
+ successfully initialized with inflateBackInit(). The input file may consist
+ of a series of gzip streams, in which case all of them will be decompressed
+ to the output file. If outfile is -1, then the gzip stream(s) integrity is
+ checked and nothing is written.
+
+ The return value is a zlib error code: Z_MEM_ERROR if out of memory,
+ Z_DATA_ERROR if the header or the compressed data is invalid, or if the
+ trailer CRC-32 check or length doesn't match, Z_BUF_ERROR if the input ends
+ prematurely or a write error occurs, or Z_ERRNO if junk (not a another gzip
+ stream) follows a valid gzip stream.
+ */
+local int gunpipe(z_stream *strm, int infile, int outfile)
+{
+ int ret, first, last;
+ unsigned have, flags, len;
+ z_const unsigned char *next = NULL;
+ struct ind ind, *indp;
+ struct outd outd;
+
+ /* setup input buffer */
+ ind.infile = infile;
+ ind.inbuf = inbuf;
+ indp = &ind;
+
+ /* decompress concatenated gzip streams */
+ have = 0; /* no input data read in yet */
+ first = 1; /* looking for first gzip header */
+ strm->next_in = Z_NULL; /* so Z_BUF_ERROR means EOF */
+ for (;;) {
+ /* look for the two magic header bytes for a gzip stream */
+ if (NEXT() == -1) {
+ ret = Z_OK;
+ break; /* empty gzip stream is ok */
+ }
+ if (last != 31 || (NEXT() != 139 && last != 157)) {
+ strm->msg = (char *)"incorrect header check";
+ ret = first ? Z_DATA_ERROR : Z_ERRNO;
+ break; /* not a gzip or compress header */
+ }
+ first = 0; /* next non-header is junk */
+
+ /* process a compress (LZW) file -- can't be concatenated after this */
+ if (last == 157) {
+ ret = lunpipe(have, next, indp, outfile, strm);
+ break;
+ }
+
+ /* process remainder of gzip header */
+ ret = Z_BUF_ERROR;
+ if (NEXT() != 8) { /* only deflate method allowed */
+ if (last == -1) break;
+ strm->msg = (char *)"unknown compression method";
+ ret = Z_DATA_ERROR;
+ break;
+ }
+ flags = NEXT(); /* header flags */
+ NEXT(); /* discard mod time, xflgs, os */
+ NEXT();
+ NEXT();
+ NEXT();
+ NEXT();
+ NEXT();
+ if (last == -1) break;
+ if (flags & 0xe0) {
+ strm->msg = (char *)"unknown header flags set";
+ ret = Z_DATA_ERROR;
+ break;
+ }
+ if (flags & 4) { /* extra field */
+ len = NEXT();
+ len += (unsigned)(NEXT()) << 8;
+ if (last == -1) break;
+ while (len > have) {
+ len -= have;
+ have = 0;
+ if (NEXT() == -1) break;
+ len--;
+ }
+ if (last == -1) break;
+ have -= len;
+ next += len;
+ }
+ if (flags & 8) /* file name */
+ while (NEXT() != 0 && last != -1)
+ ;
+ if (flags & 16) /* comment */
+ while (NEXT() != 0 && last != -1)
+ ;
+ if (flags & 2) { /* header crc */
+ NEXT();
+ NEXT();
+ }
+ if (last == -1) break;
+
+ /* set up output */
+ outd.outfile = outfile;
+ outd.check = 1;
+ outd.crc = crc32(0L, Z_NULL, 0);
+ outd.total = 0;
+
+ /* decompress data to output */
+ strm->next_in = next;
+ strm->avail_in = have;
+ ret = inflateBack(strm, in, indp, out, &outd);
+ if (ret != Z_STREAM_END) break;
+ next = strm->next_in;
+ have = strm->avail_in;
+ strm->next_in = Z_NULL; /* so Z_BUF_ERROR means EOF */
+
+ /* check trailer */
+ ret = Z_BUF_ERROR;
+ if (NEXT() != (int)(outd.crc & 0xff) ||
+ NEXT() != (int)((outd.crc >> 8) & 0xff) ||
+ NEXT() != (int)((outd.crc >> 16) & 0xff) ||
+ NEXT() != (int)((outd.crc >> 24) & 0xff)) {
+ /* crc error */
+ if (last != -1) {
+ strm->msg = (char *)"incorrect data check";
+ ret = Z_DATA_ERROR;
+ }
+ break;
+ }
+ if (NEXT() != (int)(outd.total & 0xff) ||
+ NEXT() != (int)((outd.total >> 8) & 0xff) ||
+ NEXT() != (int)((outd.total >> 16) & 0xff) ||
+ NEXT() != (int)((outd.total >> 24) & 0xff)) {
+ /* length error */
+ if (last != -1) {
+ strm->msg = (char *)"incorrect length check";
+ ret = Z_DATA_ERROR;
+ }
+ break;
+ }
+
+ /* go back and look for another gzip stream */
+ }
+
+ /* clean up and return */
+ return ret;
+}
+
+/* Copy file attributes, from -> to, as best we can. This is best effort, so
+ no errors are reported. The mode bits, including suid, sgid, and the sticky
+ bit are copied (if allowed), the owner's user id and group id are copied
+ (again if allowed), and the access and modify times are copied. */
+local void copymeta(char *from, char *to)
+{
+ struct stat was;
+ struct utimbuf when;
+
+ /* get all of from's Unix meta data, return if not a regular file */
+ if (stat(from, &was) != 0 || (was.st_mode & S_IFMT) != S_IFREG)
+ return;
+
+ /* set to's mode bits, ignore errors */
+ (void)chmod(to, was.st_mode & 07777);
+
+ /* copy owner's user and group, ignore errors */
+ (void)chown(to, was.st_uid, was.st_gid);
+
+ /* copy access and modify times, ignore errors */
+ when.actime = was.st_atime;
+ when.modtime = was.st_mtime;
+ (void)utime(to, &when);
+}
+
+/* Decompress the file inname to the file outnname, of if test is true, just
+ decompress without writing and check the gzip trailer for integrity. If
+ inname is NULL or an empty string, read from stdin. If outname is NULL or
+ an empty string, write to stdout. strm is a pre-initialized inflateBack
+ structure. When appropriate, copy the file attributes from inname to
+ outname.
+
+ gunzip() returns 1 if there is an out-of-memory error or an unexpected
+ return code from gunpipe(). Otherwise it returns 0.
+ */
+local int gunzip(z_stream *strm, char *inname, char *outname, int test)
+{
+ int ret;
+ int infile, outfile;
+
+ /* open files */
+ if (inname == NULL || *inname == 0) {
+ inname = "-";
+ infile = 0; /* stdin */
+ }
+ else {
+ infile = open(inname, O_RDONLY, 0);
+ if (infile == -1) {
+ fprintf(stderr, "gun cannot open %s\n", inname);
+ return 0;
+ }
+ }
+ if (test)
+ outfile = -1;
+ else if (outname == NULL || *outname == 0) {
+ outname = "-";
+ outfile = 1; /* stdout */
+ }
+ else {
+ outfile = open(outname, O_CREAT | O_TRUNC | O_WRONLY, 0666);
+ if (outfile == -1) {
+ close(infile);
+ fprintf(stderr, "gun cannot create %s\n", outname);
+ return 0;
+ }
+ }
+ errno = 0;
+
+ /* decompress */
+ ret = gunpipe(strm, infile, outfile);
+ if (outfile > 2) close(outfile);
+ if (infile > 2) close(infile);
+
+ /* interpret result */
+ switch (ret) {
+ case Z_OK:
+ case Z_ERRNO:
+ if (infile > 2 && outfile > 2) {
+ copymeta(inname, outname); /* copy attributes */
+ unlink(inname);
+ }
+ if (ret == Z_ERRNO)
+ fprintf(stderr, "gun warning: trailing garbage ignored in %s\n",
+ inname);
+ break;
+ case Z_DATA_ERROR:
+ if (outfile > 2) unlink(outname);
+ fprintf(stderr, "gun data error on %s: %s\n", inname, strm->msg);
+ break;
+ case Z_MEM_ERROR:
+ if (outfile > 2) unlink(outname);
+ fprintf(stderr, "gun out of memory error--aborting\n");
+ return 1;
+ case Z_BUF_ERROR:
+ if (outfile > 2) unlink(outname);
+ if (strm->next_in != Z_NULL) {
+ fprintf(stderr, "gun write error on %s: %s\n",
+ outname, strerror(errno));
+ }
+ else if (errno) {
+ fprintf(stderr, "gun read error on %s: %s\n",
+ inname, strerror(errno));
+ }
+ else {
+ fprintf(stderr, "gun unexpected end of file on %s\n",
+ inname);
+ }
+ break;
+ default:
+ if (outfile > 2) unlink(outname);
+ fprintf(stderr, "gun internal error--aborting\n");
+ return 1;
+ }
+ return 0;
+}
+
+/* Process the gun command line arguments. See the command syntax near the
+ beginning of this source file. */
+int main(int argc, char **argv)
+{
+ int ret, len, test;
+ char *outname;
+ unsigned char *window;
+ z_stream strm;
+
+ /* initialize inflateBack state for repeated use */
+ window = match; /* reuse LZW match buffer */
+ strm.zalloc = Z_NULL;
+ strm.zfree = Z_NULL;
+ strm.opaque = Z_NULL;
+ ret = inflateBackInit(&strm, 15, window);
+ if (ret != Z_OK) {
+ fprintf(stderr, "gun out of memory error--aborting\n");
+ return 1;
+ }
+
+ /* decompress each file to the same name with the suffix removed */
+ argc--;
+ argv++;
+ test = 0;
+ if (argc && strcmp(*argv, "-h") == 0) {
+ fprintf(stderr, "gun 1.6 (17 Jan 2010)\n");
+ fprintf(stderr, "Copyright (C) 2003-2010 Mark Adler\n");
+ fprintf(stderr, "usage: gun [-t] [file1.gz [file2.Z ...]]\n");
+ return 0;
+ }
+ if (argc && strcmp(*argv, "-t") == 0) {
+ test = 1;
+ argc--;
+ argv++;
+ }
+ if (argc)
+ do {
+ if (test)
+ outname = NULL;
+ else {
+ len = (int)strlen(*argv);
+ if (strcmp(*argv + len - 3, ".gz") == 0 ||
+ strcmp(*argv + len - 3, "-gz") == 0)
+ len -= 3;
+ else if (strcmp(*argv + len - 2, ".z") == 0 ||
+ strcmp(*argv + len - 2, "-z") == 0 ||
+ strcmp(*argv + len - 2, "_z") == 0 ||
+ strcmp(*argv + len - 2, ".Z") == 0)
+ len -= 2;
+ else {
+ fprintf(stderr, "gun error: no gz type on %s--skipping\n",
+ *argv);
+ continue;
+ }
+ outname = malloc(len + 1);
+ if (outname == NULL) {
+ fprintf(stderr, "gun out of memory error--aborting\n");
+ ret = 1;
+ break;
+ }
+ memcpy(outname, *argv, len);
+ outname[len] = 0;
+ }
+ ret = gunzip(&strm, *argv, outname, test);
+ if (outname != NULL) free(outname);
+ if (ret) break;
+ } while (argv++, --argc);
+ else
+ ret = gunzip(&strm, NULL, NULL, test);
+
+ /* clean up */
+ inflateBackEnd(&strm);
+ return ret;
+}
ata = self.file.read(min(size-read, 4096)) if not data: break read += len(data) chunks.append(data) return b''.join(chunks) def readline(self): """Read line from remote.""" return self.file.readline() def send(self, data): """Send data to remote.""" self.sock.sendall(data) def shutdown(self): """Close I/O established in "open".""" self.file.close() try: self.sock.shutdown(socket.SHUT_RDWR) except socket.error as e: # The server might already have closed the connection if e.errno != errno.ENOTCONN: raise finally: self.sock.close() def socket(self): """Return socket instance used to connect to IMAP4 server. socket = <instance>.socket() """ return self.sock # Utility methods def recent(self): """Return most recent 'RECENT' responses if any exist, else prompt server for an update using the 'NOOP' command. (typ, [data]) = <instance>.recent() 'data' is None if no new messages, else list of RECENT responses, most recent last. """ name = 'RECENT' typ, dat = self._untagged_response('OK', [None], name) if dat[-1]: return typ, dat typ, dat = self.noop() # Prod server for response return self._untagged_response(typ, dat, name) def response(self, code): """Return data for response 'code' if received, or None. Old value for response 'code' is cleared. (code, [data]) = <instance>.response(code) """ return self._untagged_response(code, [None], code.upper()) # IMAP4 commands def append(self, mailbox, flags, date_time, message): """Append message to named mailbox. (typ, [data]) = <instance>.append(mailbox, flags, date_time, message) All args except `message' can be None. """ name = 'APPEND' if not mailbox: mailbox = 'INBOX' if flags: if (flags[0],flags[-1]) != ('(',')'): flags = '(%s)' % flags else: flags = None if date_time: date_time = Time2Internaldate(date_time) else: date_time = None self.literal = MapCRLF.sub(CRLF, message) return self._simple_command(name, mailbox, flags, date_time) def authenticate(self, mechanism, authobject): """Authenticate command - requires response processing. 'mechanism' specifies which authentication mechanism is to be used - it must appear in <instance>.capabilities in the form AUTH=<mechanism>. 'authobject' must be a callable object: data = authobject(response) It will be called to process server continuation responses; the response argument it is passed will be a bytes. It should return bytes data that will be base64 encoded and sent to the server. It should return None if the client abort response '*' should be sent instead. """ mech = mechanism.upper() # XXX: shouldn't this code be removed, not commented out? #cap = 'AUTH=%s' % mech #if not cap in self.capabilities: # Let the server decide! # raise self.error("Server doesn't allow %s authentication." % mech) self.literal = _Authenticator(authobject).process typ, dat = self._simple_command('AUTHENTICATE', mech) if typ != 'OK': raise self.error(dat[-1]) self.state = 'AUTH' return typ, dat def capability(self): """(typ, [data]) = <instance>.capability() Fetch capabilities list from server.""" name = 'CAPABILITY' typ, dat = self._simple_command(name) return self._untagged_response(typ, dat, name) def check(self): """Checkpoint mailbox on server. (typ, [data]) = <instance>.check() """ return self._simple_command('CHECK') def close(self): """Close currently selected mailbox. Deleted messages are removed from writable mailbox. This is the recommended command before 'LOGOUT'. (typ, [data]) = <instance>.close() """ try: typ, dat = self._simple_command('CLOSE') finally: self.state = 'AUTH' return typ, dat def copy(self, message_set, new_mailbox): """Copy 'message_set' messages onto end of 'new_mailbox'. (typ, [data]) = <instance>.copy(message_set, new_mailbox) """ return self._simple_command('COPY', message_set, new_mailbox) def create(self, mailbox): """Create new mailbox. (typ, [data]) = <instance>.create(mailbox) """ return self._simple_command('CREATE', mailbox) def delete(self, mailbox): """Delete old mailbox. (typ, [data]) = <instance>.delete(mailbox) """ return self._simple_command('DELETE', mailbox) def deleteacl(self, mailbox, who): """Delete the ACLs (remove any rights) set for who on mailbox. (typ, [data]) = <instance>.deleteacl(mailbox, who) """ return self._simple_command('DELETEACL', mailbox, who) def expunge(self): """Permanently remove deleted items from selected mailbox. Generates 'EXPUNGE' response for each deleted message. (typ, [data]) = <instance>.expunge() 'data' is list of 'EXPUNGE'd message numbers in order received. """ name = 'EXPUNGE' typ, dat = self._simple_command(name) return self._untagged_response(typ, dat, name) def fetch(self, message_set, message_parts): """Fetch (parts of) messages. (typ, [data, ...]) = <instance>.fetch(message_set, message_parts) 'message_parts' should be a string of selected parts enclosed in parentheses, eg: "(UID BODY[TEXT])". 'data' are tuples of message part envelope and data. """ name = 'FETCH' typ, dat = self._simple_command(name, message_set, message_parts) return self._untagged_response(typ, dat, name) def getacl(self, mailbox): """Get the ACLs for a mailbox. (typ, [data]) = <instance>.getacl(mailbox) """ typ, dat = self._simple_command('GETACL', mailbox) return self._untagged_response(typ, dat, 'ACL') def getannotation(self, mailbox, entry, attribute): """(typ, [data]) = <instance>.getannotation(mailbox, entry, attribute) Retrieve ANNOTATIONs.""" typ, dat = self._simple_command('GETANNOTATION', mailbox, entry, attribute) return self._untagged_response(typ, dat, 'ANNOTATION') def getquota(self, root): """Get the quota root's resource usage and limits. Part of the IMAP4 QUOTA extension defined in rfc2087. (typ, [data]) = <instance>.getquota(root) """ typ, dat = self._simple_command('GETQUOTA', root) return self._untagged_response(typ, dat, 'QUOTA') def getquotaroot(self, mailbox): """Get the list of quota roots for the named mailbox. (typ, [[QUOTAROOT responses...], [QUOTA responses]]) = <instance>.getquotaroot(mailbox) """ typ, dat = self._simple_command('GETQUOTAROOT', mailbox) typ, quota = self._untagged_response(typ, dat, 'QUOTA') typ, quotaroot = self._untagged_response(typ, dat, 'QUOTAROOT') return typ, [quotaroot, quota] def list(self, directory='""', pattern='*'): """List mailbox names in directory matching pattern. (typ, [data]) = <instance>.list(directory='""', pattern='*') 'data' is list of LIST responses. """ name = 'LIST' typ, dat = self._simple_command(name, directory, pattern) return self._untagged_response(typ, dat, name) def login(self, user, password): """Identify client using plaintext password. (typ, [data]) = <instance>.login(user, password) NB: 'password' will be quoted. """ typ, dat = self._simple_command('LOGIN', user, self._quote(password)) if typ != 'OK': raise self.error(dat[-1]) self.state = 'AUTH' return typ, dat def login_cram_md5(self, user, password): """ Force use of CRAM-MD5 authentication. (typ, [data]) = <instance>.login_cram_md5(user, password) """ self.user, self.password = user, password return self.authenticate('CRAM-MD5', self._CRAM_MD5_AUTH) def _CRAM_MD5_AUTH(self, challenge): """ Authobject to use with CRAM-MD5 authentication. """ import hmac pwd = (self.password.encode('ASCII') if isinstance(self.password, str) else self.password) return self.user + " " + hmac.HMAC(pwd, challenge).hexdigest() def logout(self): """Shutdown connection to server. (typ, [data]) = <instance>.logout() Returns server 'BYE' response. """ self.state = 'LOGOUT' try: typ, dat = self._simple_command('LOGOUT') except: typ, dat = 'NO', ['%s: %s' % sys.exc_info()[:2]] self.shutdown() if 'BYE' in self.untagged_responses: return 'BYE', self.untagged_responses['BYE'] return typ, dat def lsub(self, directory='""', pattern='*'): """List 'subscribed' mailbox names in directory matching pattern. (typ, [data, ...]) = <instance>.lsub(directory='""', pattern='*') 'data' are tuples of message part envelope and data. """ name = 'LSUB' typ, dat = self._simple_command(name, directory, pattern) return self._untagged_response(typ, dat, name) def myrights(self, mailbox): """Show my ACLs for a mailbox (i.e. the rights that I have on mailbox). (typ, [data]) = <instance>.myrights(mailbox) """ typ,dat = self._simple_command('MYRIGHTS', mailbox) return self._untagged_response(typ, dat, 'MYRIGHTS') def namespace(self): """ Returns IMAP namespaces ala rfc2342 (typ, [data, ...]) = <instance>.namespace() """ name = 'NAMESPACE' typ, dat = self._simple_command(name) return self._untagged_response(typ, dat, name) def noop(self): """Send NOOP command. (typ, [data]) = <instance>.noop() """ if __debug__: if self.debug >= 3: self._dump_ur(self.untagged_responses) return self._simple_command('NOOP') def partial(self, message_num, message_part, start, length): """Fetch truncated part of a message. (typ, [data, ...]) = <instance>.partial(message_num, message_part, start, length) 'data' is tuple of message part envelope and data. """ name = 'PARTIAL' typ, dat = self._simple_command(name, message_num, message_part, start, length) return self._untagged_response(typ, dat, 'FETCH') def proxyauth(self, user): """Assume authentication as "user". Allows an authorised administrator to proxy into any user's mailbox. (typ, [data]) = <instance>.proxyauth(user) """ name = 'PROXYAUTH' return self._simple_command('PROXYAUTH', user) def rename(self, oldmailbox, newmailbox): """Rename old mailbox name to new. (typ, [data]) = <instance>.rename(oldmailbox, newmailbox) """ return self._simple_command('RENAME', oldmailbox, newmailbox) def search(self, charset, *criteria): """Search mailbox for matching messages. (typ, [data]) = <instance>.search(charset, criterion, ...) 'data' is space separated list of matching message numbers. """ name = 'SEARCH' if charset: typ, dat = self._simple_command(name, 'CHARSET', charset, *criteria) else: typ, dat = self._simple_command(name, *criteria) return self._untagged_response(typ, dat, name) def select(self, mailbox='INBOX', readonly=False): """Select a mailbox. Flush all untagged responses. (typ, [data]) = <instance>.select(mailbox='INBOX', readonly=False) 'data' is count of messages in mailbox ('EXISTS' response). Mandated responses are ('FLAGS', 'EXISTS', 'RECENT', 'UIDVALIDITY'), so other responses should be obtained via <instance>.response('FLAGS') etc. """ self.untagged_responses = {} # Flush old responses. self.is_readonly = readonly if readonly: name = 'EXAMINE' else: name = 'SELECT' typ, dat = self._simple_command(name, mailbox) if typ != 'OK': self.state = 'AUTH' # Might have been 'SELECTED' return typ, dat self.state = 'SELECTED' if 'READ-ONLY' in self.untagged_responses \ and not readonly: if __debug__: if self.debug >= 1: self._dump_ur(self.untagged_responses) raise self.readonly('%s is not writable' % mailbox) return typ, self.untagged_responses.get('EXISTS', [None]) def setacl(self, mailbox, who, what): """Set a mailbox acl. (typ, [data]) = <instance>.setacl(mailbox, who, what) """ return self._simple_command('SETACL', mailbox, who, what) def setannotation(self, *args): """(typ, [data]) = <instance>.setannotation(mailbox[, entry, attribute]+) Set ANNOTATIONs.""" typ, dat = self._simple_command('SETANNOTATION', *args) return self._untagged_response(typ, dat, 'ANNOTATION') def setquota(self, root, limits): """Set the quota root's resource limits. (typ, [data]) = <instance>.setquota(root, limits) """ typ, dat = self._simple_command('SETQUOTA', root, limits) return self._untagged_response(typ, dat, 'QUOTA') def sort(self, sort_criteria, charset, *search_criteria): """IMAP4rev1 extension SORT command. (typ, [data]) = <instance>.sort(sort_criteria, charset, search_criteria, ...) """ name = 'SORT' #if not name in self.capabilities: # Let the server decide! # raise self.error('unimplemented extension command: %s' % name) if (sort_criteria[0],sort_criteria[-1]) != ('(',')'): sort_criteria = '(%s)' % sort_criteria typ, dat = self._simple_command(name, sort_criteria, charset, *search_criteria) return self._untagged_response(typ, dat, name) def starttls(self, ssl_context=None): name = 'STARTTLS' if not HAVE_SSL: raise self.error('SSL support missing') if self._tls_established: raise self.abort('TLS session already established') if name not in self.capabilities: raise self.abort('TLS not supported by server') # Generate a default SSL context if none was passed. if ssl_context is None: ssl_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23) # SSLv2 considered harmful. ssl_context.options |= ssl.OP_NO_SSLv2 typ, dat = self._simple_command(name) if typ == 'OK': self.sock = ssl_context.wrap_socket(self.sock) self.file = self.sock.makefile('rb') self._tls_established = True self._get_capabilities() else: raise self.error("Couldn't establish TLS session") return self._untagged_response(typ, dat, name) def status(self, mailbox, names): """Request named status conditions for mailbox. (typ, [data]) = <instance>.status(mailbox, names) """ name = 'STATUS' #if self.PROTOCOL_VERSION == 'IMAP4': # Let the server decide! # raise self.error('%s unimplemented in IMAP4 (obtain IMAP4rev1 server, or re-code)' % name) typ, dat = self._simple_command(name, mailbox, names) return self._untagged_response(typ, dat, name) def store(self, message_set, command, flags): """Alters flag dispositions for messages in mailbox. (typ, [data]) = <instance>.store(message_set, command, flags) """ if (flags[0],flags[-1]) != ('(',')'): flags = '(%s)' % flags # Avoid quoting the flags typ, dat = self._simple_command('STORE', message_set, command, flags) return self._untagged_response(typ, dat, 'FETCH') def subscribe(self, mailbox): """Subscribe to new mailbox. (typ, [data]) = <instance>.subscribe(mailbox) """ return self._simple_command('SUBSCRIBE', mailbox) def thread(self, threading_algorithm, charset, *search_criteria): """IMAPrev1 extension THREAD command. (type, [data]) = <instance>.thread(threading_algorithm, charset, search_criteria, ...) """ name = 'THREAD' typ, dat = self._simple_command(name, threading_algorithm, charset, *search_criteria) return self._untagged_response(typ, dat, name) def uid(self, command, *args): """Execute "command arg ..." with messages identified by UID, rather than message number. (typ, [data]) = <instance>.uid(command, arg1, arg2, ...) Returns response appropriate to 'command'. """ command = command.upper() if not command in Commands: raise self.error("Unknown IMAP4 UID command: %s" % command) if self.state not in Commands[command]: raise self.error("command %s illegal in state %s, " "only allowed in states %s" % (command, self.state, ', '.join(Commands[command]))) name = 'UID' typ, dat = self._simple_command(name, command, *args) if command in ('SEARCH', 'SORT', 'THREAD'): name = command else: name = 'FETCH' return self._untagged_response(typ, dat, name) def unsubscribe(self, mailbox): """Unsubscribe from old mailbox. (typ, [data]) = <instance>.unsubscribe(mailbox) """ return self._simple_command('UNSUBSCRIBE', mailbox) def xatom(self, name, *args): """Allow simple extension commands notified by server in CAPABILITY response. Assumes command is legal in current state. (typ, [data]) = <instance>.xatom(name, arg, ...) Returns response appropriate to extension command `name'. """ name = name.upper() #if not name in self.capabilities: # Let the server decide! # raise self.error('unknown extension command: %s' % name) if not name in Commands: Commands[name] = (self.state,) return self._simple_command(name, *args) # Private methods def _append_untagged(self, typ, dat): if dat is None: dat = b'' ur = self.untagged_responses if __debug__: if self.debug >= 5: self._mesg('untagged_responses[%s] %s += ["%r"]' % (typ, len(ur.get(typ,'')), dat)) if typ in ur: ur[typ].append(dat) else: ur[typ] = [dat] def _check_bye(self): bye = self.untagged_responses.get('BYE') if bye: raise self.abort(bye[-1].decode('ascii', 'replace')) def _command(self, name, *args): if self.state not in Commands[name]: self.literal = None raise self.error("command %s illegal in state %s, " "only allowed in states %s" % (name, self.state, ', '.join(Commands[name]))) for typ in ('OK', 'NO', 'BAD'): if typ in self.untagged_responses: del self.untagged_responses[typ] if 'READ-ONLY' in self.untagged_responses \ and not self.is_readonly: raise self.readonly('mailbox status changed to READ-ONLY') tag = self._new_tag() name = bytes(name, 'ASCII') data = tag + b' ' + name for arg in args: if arg is None: continue if isinstance(arg, str): arg = bytes(arg, "ASCII") data = data + b' ' + arg literal = self.literal if literal is not None: self.literal = None if type(literal) is type(self._command): literator = literal else: literator = None data = data + bytes(' {%s}' % len(literal), 'ASCII') if __debug__: if self.debug >= 4: self._mesg('> %r' % data) else: self._log('> %r' % data) try: self.send(data + CRLF) except (socket.error, OSError) as val: raise self.abort('socket error: %s' % val) if literal is None: return tag while 1: # Wait for continuation response while self._get_response(): if self.tagged_commands[tag]: # BAD/NO? return tag # Send literal if literator: literal = literator(self.continuation_response) if __debug__: if self.debug >= 4: self._mesg('write literal size %s' % len(literal)) try: self.send(literal) self.send(CRLF) except (socket.error, OSError) as val: raise self.abort('socket error: %s' % val) if not literator: break return tag def _command_complete(self, name, tag): # BYE is expected after LOGOUT if name != 'LOGOUT': self._check_bye() try: typ, data = self._get_tagged_response(tag) except self.abort as val: raise self.abort('command: %s => %s' % (name, val)) except self.error as val: raise self.error('command: %s => %s' % (name, val)) if name != 'LOGOUT': self._check_bye() if typ == 'BAD': raise self.error('%s command error: %s %s' % (name, typ, data)) return typ, data def _get_capabilities(self): typ, dat = self.capability() if dat == [None]: raise self.error('no CAPABILITY response from server') dat = str(dat[-1], "ASCII") dat = dat.upper() self.capabilities = tuple(dat.split()) def _get_response(self): # Read response and store. # # Returns None for continuation responses, # otherwise first response line received. resp = self._get_line() # Command completion response? if self._match(self.tagre, resp): tag = self.mo.group('tag') if not tag in self.tagged_commands: raise self.abort('unexpected tagged response: %s' % resp) typ = self.mo.group('type') typ = str(typ, 'ASCII') dat = self.mo.group('data') self.tagged_commands[tag] = (typ, [dat]) else: dat2 = None # '*' (untagged) responses? if not self._match(Untagged_response, resp): if self._match(Untagged_status, resp): dat2 = self.mo.group('data2') if self.mo is None: # Only other possibility is '+' (continuation) response... if self._match(Continuation, resp): self.continuation_response = self.mo.group('data') return None # NB: indicates continuation raise self.abort("unexpected response: '%s'" % resp) typ = self.mo.group('type') typ = str(typ, 'ascii') dat = self.mo.group('data') if dat is None: dat = b'' # Null untagged response if dat2: dat = dat + b' ' + dat2 # Is there a literal to come? while self._match(Literal, dat): # Read literal direct from connection. size = int(self.mo.group('size')) if __debug__: if self.debug >= 4: self._mesg('read literal size %s' % size) data = self.read(size) # Store response with literal as tuple self._append_untagged(typ, (dat, data)) # Read trailer - possibly containing another literal dat = self._get_line() self._append_untagged(typ, dat) # Bracketed response information? if typ in ('OK', 'NO', 'BAD') and self._match(Response_code, dat): typ = self.mo.group('type') typ = str(typ, "ASCII") self._append_untagged(typ, self.mo.group('data')) if __debug__: if self.debug >= 1 and typ in ('NO', 'BAD', 'BYE'): self._mesg('%s response: %r' % (typ, dat)) return resp def _get_tagged_response(self, tag): while 1: result = self.tagged_commands[tag] if result is not None: del self.tagged_commands[tag] return result # Some have reported "unexpected response" exceptions. # Note that ignoring them here causes loops. # Instead, send me details of the unexpected response and # I'll update the code in `_get_response()'. try: self._get_response() except self.abort as val: if __debug__: if self.debug >= 1: self.print_log() raise def _get_line(self): line = self.readline() if not line: raise self.abort('socket error: EOF') # Protocol mandates all lines terminated by CRLF if not line.endswith(b'\r\n'): raise self.abort('socket error: unterminated line') line = line[:-2] if __debug__: if self.debug >= 4: self._mesg('< %r' % line) else: self._log('< %r' % line) return line def _match(self, cre, s): # Run compiled regular expression match method on 's'. # Save result, return success. self.mo = cre.match(s) if __debug__: if self.mo is not None and self.debug >= 5: self._mesg("\tmatched r'%r' => %r" % (cre.pattern, self.mo.groups())) return self.mo is not None def _new_tag(self): tag = self.tagpre + bytes(str(self.tagnum), 'ASCII') self.tagnum = self.tagnum + 1 self.tagged_commands[tag] = None return tag def _quote(self, arg): arg = arg.replace('\\', '\\\\') arg = arg.replace('"', '\\"') return '"' + arg + '"' def _simple_command(self, name, *args): return self._command_complete(name, self._command(name, *args)) def _untagged_response(self, typ, dat, name): if typ == 'NO': return typ, dat if not name in self.untagged_responses: return typ, [None] data = self.untagged_responses.pop(name) if __debug__: if self.debug >= 5: self._mesg('untagged_responses[%s] => %s' % (name, data)) return typ, data if __debug__: def _mesg(self, s, secs=None): if secs is None: secs = time.time() tm = time.strftime('%M:%S', time.localtime(secs)) sys.stderr.write(' %s.%02d %s\n' % (tm, (secs*100)%100, s)) sys.stderr.flush() def _dump_ur(self, dict): # Dump untagged responses (in `dict'). l = dict.items() if not l: return t = '\n\t\t' l = map(lambda x:'%s: "%s"' % (x[0], x[1][0] and '" "'.join(x[1]) or ''), l) self._mesg('untagged responses dump:%s%s' % (t, t.join(l))) def _log(self, line): # Keep log of last `_cmd_log_len' interactions for debugging. self._cmd_log[self._cmd_log_idx] = (line, time.time()) self._cmd_log_idx += 1 if self._cmd_log_idx >= self._cmd_log_len: self._cmd_log_idx = 0 def print_log(self): self._mesg('last %d IMAP4 interactions:' % len(self._cmd_log)) i, n = self._cmd_log_idx, self._cmd_log_len while n: try: self._mesg(*self._cmd_log[i]) except: pass i += 1 if i >= self._cmd_log_len: i = 0 n -= 1 if HAVE_SSL: class IMAP4_SSL(IMAP4): """IMAP4 client class over SSL connection Instantiate with: IMAP4_SSL([host[, port[, keyfile[, certfile]]]]) host - host's name (default: localhost); port - port number (default: standard IMAP4 SSL port). keyfile - PEM formatted file that contains your private key (default: None); certfile - PEM formatted certificate chain file (default: None); for more documentation see the docstring of the parent class IMAP4. """ def __init__(self, host = '', port = IMAP4_SSL_PORT, keyfile = None, certfile = None): self.keyfile = keyfile self.certfile = certfile IMAP4.__init__(self, host, port) def _create_socket(self): sock = IMAP4._create_socket(self) return ssl.wrap_socket(sock, self.keyfile, self.certfile) def open(self, host='', port=IMAP4_SSL_PORT): """Setup connection to remote server on "host:port". (default: localhost:standard IMAP4 SSL port). This connection will be used by the routines: read, readline, send, shutdown. """ IMAP4.open(self, host, port) __all__.append("IMAP4_SSL") class IMAP4_stream(IMAP4): """IMAP4 client class over a stream Instantiate with: IMAP4_stream(command) where "command" is a string that can be passed to subprocess.Popen() for more documentation see the docstring of the parent class IMAP4. """ def __init__(self, command): self.command = command IMAP4.__init__(self) def open(self, host = None, port = None): """Setup a stream connection. This connection will be used by the routines: read, readline, send, shutdown. """ self.host = None # For compatibility with parent class self.port = None self.sock = None self.file = None self.process = subprocess.Popen(self.command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, shell=True, close_fds=True) self.writefile = self.process.stdin self.readfile = self.process.stdout def read(self, size): """Read 'size' bytes from remote.""" return self.readfile.read(size) def readline(self): """Read line from remote.""" return self.readfile.readline() def send(self, data): """Send data to remote.""" self.writefile.write(data) self.writefile.flush() def shutdown(self): """Close I/O established in "open".""" self.readfile.close() self.writefile.close() self.process.wait() class _Authenticator: """Private class to provide en/decoding for base64-based authentication conversation. """ def __init__(self, mechinst): self.mech = mechinst # Callable object to provide/process data def process(self, data): ret = self.mech(self.decode(data)) if ret is None: return '*' # Abort conversation return self.encode(ret) def encode(self, inp): # # Invoke binascii.b2a_base64 iteratively with # short even length buffers, strip the trailing # line feed from the result and append. "Even" # means a number that factors to both 6 and 8, # so when it gets to the end of the 8-bit input # there's no partial 6-bit output. # oup = b'' if isinstance(inp, str): inp = inp.encode('ASCII') while inp: if len(inp) > 48: t = inp[:48] inp = inp[48:] else: t = inp inp = b'' e = binascii.b2a_base64(t) if e: oup = oup + e[:-1] return oup def decode(self, inp): if not inp: return b'' return binascii.a2b_base64(inp) Mon2num = {b'Jan': 1, b'Feb': 2, b'Mar': 3, b'Apr': 4, b'May': 5, b'Jun': 6, b'Jul': 7, b'Aug': 8, b'Sep': 9, b'Oct': 10, b'Nov': 11, b'Dec': 12} def Internaldate2tuple(resp): """Parse an IMAP4 INTERNALDATE string. Return corresponding local time. The return value is a time.struct_time tuple or None if the string has wrong format. """ mo = InternalDate.match(resp) if not mo: return None mon = Mon2num[mo.group('mon')] zonen = mo.group('zonen') day = int(mo.group('day')) year = int(mo.group('year')) hour = int(mo.group('hour')) min = int(mo.group('min')) sec = int(mo.group('sec')) zoneh = int(mo.group('zoneh')) zonem = int(mo.group('zonem')) # INTERNALDATE timezone must be subtracted to get UT zone = (zoneh*60 + zonem)*60 if zonen == b'-': zone = -zone tt = (year, mon, day, hour, min, sec, -1, -1, -1) utc = calendar.timegm(tt) - zone return time.localtime(utc) def Int2AP(num): """Convert integer to A-P string representation.""" val = b''; AP = b'ABCDEFGHIJKLMNOP' num = int(abs(num)) while num: num, mod = divmod(num, 16) val = AP[mod:mod+1] + val return val def ParseFlags(resp): """Convert IMAP4 flags response to python tuple.""" mo = Flags.match(resp) if not mo: return () return tuple(mo.group('flags').split()) def Time2Internaldate(date_time): """Convert date_time to IMAP4 INTERNALDATE representation. Return string in form: '"DD-Mmm-YYYY HH:MM:SS +HHMM"'. The date_time argument can be a number (int or float) representing seconds since epoch (as returned by time.time()), a 9-tuple representing local time (as returned by time.localtime()), or a double-quoted string. In the last case, it is assumed to already be in the correct format. """ if isinstance(date_time, (int, float)): tt = time.localtime(date_time) elif isinstance(date_time, (tuple, time.struct_time)): tt = date_time elif isinstance(date_time, str) and (date_time[0],date_time[-1]) == ('"','"'): return date_time # Assume in correct format else: raise ValueError("date_time not of a known type") dt = time.strftime("%d-%b-%Y %H:%M:%S", tt) if dt[0] == '0': dt = ' ' + dt[1:] if time.daylight and tt[-1]: zone = -time.altzone else: zone = -time.timezone return '"' + dt + " %+03d%02d" % divmod(zone//60, 60) + '"' if __name__ == '__main__': # To test: invoke either as 'python imaplib.py [IMAP4_server_hostname]' # or 'python imaplib.py -s "rsh IMAP4_server_hostname exec /etc/rimapd"' # to test the IMAP4_stream class import getopt, getpass try: optlist, args = getopt.getopt(sys.argv[1:], 'd:s:') except getopt.error as val: optlist, args = (), () stream_command = None for opt,val in optlist: if opt == '-d': Debug = int(val) elif opt == '-s': stream_command = val if not args: args = (stream_command,) if not args: args = ('',) host = args[0] USER = getpass.getuser() PASSWD = getpass.getpass("IMAP password for %s on %s: " % (USER, host or "localhost")) test_mesg = 'From: %(user)s@localhost%(lf)sSubject: IMAP4 test%(lf)s%(lf)sdata...%(lf)s' % {'user':USER, 'lf':'\n'} test_seq1 = ( ('login', (USER, PASSWD)), ('create', ('/tmp/xxx 1',)), ('rename', ('/tmp/xxx 1', '/tmp/yyy')), ('CREATE', ('/tmp/yyz 2',)), ('append', ('/tmp/yyz 2', None, None, test_mesg)), ('list', ('/tmp', 'yy*')), ('select', ('/tmp/yyz 2',)), ('search', (None, 'SUBJECT', 'test')), ('fetch', ('1', '(FLAGS INTERNALDATE RFC822)')), ('store', ('1', 'FLAGS', '(\Deleted)')), ('namespace', ()), ('expunge', ()), ('recent', ()), ('close', ()), ) test_seq2 = ( ('select', ()), ('response',('UIDVALIDITY',)), ('uid', ('SEARCH', 'ALL')), ('response', ('EXISTS',)), ('append', (None, None, None, test_mesg)), ('recent', ()), ('logout', ()), ) def run(cmd, args): M._mesg('%s %s' % (cmd, args)) typ, dat = getattr(M, cmd)(*args) M._mesg('%s => %s %s' % (cmd, typ, dat)) if typ == 'NO': raise dat[0] return dat try: if stream_command: M = IMAP4_stream(stream_command) else: M = IMAP4(host) if M.state == 'AUTH': test_seq1 = test_seq1[1:] # Login not needed M._mesg('PROTOCOL_VERSION = %s' % M.PROTOCOL_VERSION) M._mesg('CAPABILITIES = %r' % (M.capabilities,)) for cmd,args in test_seq1: run(cmd, args) for ml in run('list', ('/tmp/', 'yy%')): mo = re.match(r'.*"([^"]+)"$', ml) if mo: path = mo.group(1) else: path = ml.split()[-1] run('delete', (path,)) for cmd,args in test_seq2: dat = run(cmd, args) if (cmd,args) != ('uid', ('SEARCH', 'ALL')): continue uid = dat[-1].split() if not uid: continue run('uid', ('FETCH', '%s' % uid[-1], '(FLAGS INTERNALDATE RFC822.SIZE RFC822.HEADER RFC822.TEXT)')) print('\nAll tests OK.') except: print('\nTests failed.') if not Debug: print(''' If you would like to see debugging output, try: %s -d5 ''' % sys.argv[0]) raise