[thirdparty/systemd.git] / src / import / import-job.c

/*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/

/***
  This file is part of systemd.

  Copyright 2015 Lennart Poettering

  systemd is free software; you can redistribute it and/or modify it
  under the terms of the GNU Lesser General Public License as published by
  the Free Software Foundation; either version 2.1 of the License, or
  (at your option) any later version.

  systemd is distributed in the hope that it will be useful, but
  WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received a copy of the GNU Lesser General Public License
  along with systemd; If not, see <http://www.gnu.org/licenses/>.
***/

#include <sys/xattr.h>

#include "strv.h"
#include "import-job.h"

ImportJob* import_job_unref(ImportJob *j) {
        if (!j)
                return NULL;

        curl_glue_remove_and_free(j->glue, j->curl);
        curl_slist_free_all(j->request_header);

        safe_close(j->disk_fd);

        if (j->compressed == IMPORT_JOB_XZ)
                lzma_end(&j->xz);
        else if (j->compressed == IMPORT_JOB_GZIP)
                inflateEnd(&j->gzip);

        free(j->url);
        free(j->etag);
        strv_free(j->old_etags);
        free(j->payload);

        free(j);

        return NULL;
}

DEFINE_TRIVIAL_CLEANUP_FUNC(ImportJob*, import_job_unref);

static void import_job_finish(ImportJob *j, int ret) {
        assert(j);

        if (j->state == IMPORT_JOB_DONE ||
            j->state == IMPORT_JOB_FAILED)
                return;

        if (ret == 0) {
                j->state = IMPORT_JOB_DONE;
                log_info("Download of %s complete.", j->url);
        } else {
                j->state = IMPORT_JOB_FAILED;
                j->error = ret;
        }

        if (j->on_finished)
                j->on_finished(j);
}

void import_job_curl_on_finished(CurlGlue *g, CURL *curl, CURLcode result) {
        ImportJob *j = NULL;
        CURLcode code;
        long status;
        int r;

        if (curl_easy_getinfo(curl, CURLINFO_PRIVATE, &j) != CURLE_OK)
                return;

        if (!j || j->state == IMPORT_JOB_DONE || j->state == IMPORT_JOB_FAILED)
                return;

        if (result != CURLE_OK) {
                log_error("Transfer failed: %s", curl_easy_strerror(result));
                r = -EIO;
                goto finish;
        }

        code = curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &status);
        if (code != CURLE_OK) {
                log_error("Failed to retrieve response code: %s", curl_easy_strerror(code));
                r = -EIO;
                goto finish;
        } else if (status == 304) {
                log_info("Image already downloaded. Skipping download.");
                r = 0;
                goto finish;
        } else if (status >= 300) {
                log_error("HTTP request to %s failed with code %li.", j->url, status);
                r = -EIO;
                goto finish;
        } else if (status < 200) {
                log_error("HTTP request to %s finished with unexpected code %li.", j->url, status);
                r = -EIO;
                goto finish;
        }

        if (j->state != IMPORT_JOB_RUNNING) {
                log_error("Premature connection termination.");
                r = -EIO;
                goto finish;
        }

        if (j->content_length != (uint64_t) -1 &&
            j->content_length != j->written_compressed) {
                log_error("Download truncated.");
                r = -EIO;
                goto finish;
        }

        if (j->disk_fd >= 0 && j->allow_sparse) {
                /* Make sure the file size is right, in case the file was
                 * sparse and we just seeked for the last part */

                if (ftruncate(j->disk_fd, j->written_uncompressed) < 0) {
                        log_error_errno(errno, "Failed to truncate file: %m");
                        r = -errno;
                        goto finish;
                }

                if (j->etag)
                        (void) fsetxattr(j->disk_fd, "user.source_etag", j->etag, strlen(j->etag), 0);
                if (j->url)
                        (void) fsetxattr(j->disk_fd, "user.source_url", j->url, strlen(j->url), 0);

                if (j->mtime != 0) {
                        struct timespec ut[2];

                        timespec_store(&ut[0], j->mtime);
                        ut[1] = ut[0];
                        (void) futimens(j->disk_fd, ut);

                        (void) fd_setcrtime(j->disk_fd, j->mtime);
                }
        }

        r = 0;

finish:
        import_job_finish(j, r);
}


static int import_job_write_uncompressed(ImportJob *j, void *p, size_t sz) {
        ssize_t n;

        assert(j);
        assert(p);
        assert(sz > 0);
        assert(j->disk_fd >= 0);

        if (j->written_uncompressed + sz < j->written_uncompressed) {
                log_error("File too large, overflow");
                return -EOVERFLOW;
        }

        if (j->written_uncompressed + sz > j->uncompressed_max) {
                log_error("File overly large, refusing");
                return -EFBIG;
        }

        if (j->disk_fd >= 0) {

                if (j->allow_sparse)
                        n = sparse_write(j->disk_fd, p, sz, 64);
                else
                        n = write(j->disk_fd, p, sz);
                if (n < 0) {
                        log_error_errno(errno, "Failed to write file: %m");
                        return -errno;
                }
                if ((size_t) n < sz) {
                        log_error("Short write");
                        return -EIO;
                }
        } else {

                if (!GREEDY_REALLOC(j->payload, j->payload_allocated, j->payload_size + sz))
                        return log_oom();

                memcpy((uint8_t*) j->payload + j->payload_size, p, sz);
                j->payload_size += sz;
        }

        j->written_uncompressed += sz;

        return 0;
}

static int import_job_write_compressed(ImportJob *j, void *p, size_t sz) {
        int r;

        assert(j);
        assert(p);
        assert(sz > 0);
        assert(j->disk_fd >= 0);

        if (j->written_compressed + sz < j->written_compressed) {
                log_error("File too large, overflow");
                return -EOVERFLOW;
        }

        if (j->written_compressed + sz > j->compressed_max) {
                log_error("File overly large, refusing.");
                return -EFBIG;
        }

        if (j->content_length != (uint64_t) -1 &&
            j->written_compressed + sz > j->content_length) {
                log_error("Content length incorrect.");
                return -EFBIG;
        }

        switch (j->compressed) {

        case IMPORT_JOB_UNCOMPRESSED:
                r = import_job_write_uncompressed(j, p, sz);
                if (r < 0)
                        return r;

                break;

        case IMPORT_JOB_XZ:
                j->xz.next_in = p;
                j->xz.avail_in = sz;

                while (j->xz.avail_in > 0) {
                        uint8_t buffer[16 * 1024];
                        lzma_ret lzr;

                        j->xz.next_out = buffer;
                        j->xz.avail_out = sizeof(buffer);

                        lzr = lzma_code(&j->xz, LZMA_RUN);
                        if (lzr != LZMA_OK && lzr != LZMA_STREAM_END) {
                                log_error("Decompression error.");
                                return -EIO;
                        }

                        r = import_job_write_uncompressed(j, buffer, sizeof(buffer) - j->xz.avail_out);
                        if (r < 0)
                                return r;
                }

                break;

        case IMPORT_JOB_GZIP:
                j->gzip.next_in = p;
                j->gzip.avail_in = sz;

                while (j->gzip.avail_in > 0) {
                        uint8_t buffer[16 * 1024];

                        j->gzip.next_out = buffer;
                        j->gzip.avail_out = sizeof(buffer);

                        r = inflate(&j->gzip, Z_NO_FLUSH);
                        if (r != Z_OK && r != Z_STREAM_END) {
                                log_error("Decompression error.");
                                return -EIO;
                        }

                        r = import_job_write_uncompressed(j, buffer, sizeof(buffer) - j->gzip.avail_out);
                        if (r < 0)
                                return r;
                }

                break;

        default:
                assert_not_reached("Unknown compression");
        }

        j->written_compressed += sz;

        return 0;
}

static int import_job_open_disk(ImportJob *j) {
        int r;

        assert(j);

        if (j->on_open_disk) {
                r = j->on_open_disk(j);
                if (r < 0)
                        return r;
        }

        if (j->disk_fd >= 0) {
                /* Check if we can do sparse files */

                if (lseek(j->disk_fd, SEEK_SET, 0) == 0)
                        j->allow_sparse = true;
                else {
                        if (errno != ESPIPE)
                                return log_error_errno(errno, "Failed to seek on file descriptor: %m");

                        j->allow_sparse = false;
                }
        }

        return 0;
}

static int import_job_detect_compression(ImportJob *j) {
        static const uint8_t xz_signature[] = {
                0xfd, '7', 'z', 'X', 'Z', 0x00
        };
        static const uint8_t gzip_signature[] = {
                0x1f, 0x8b
        };

        _cleanup_free_ uint8_t *stub = NULL;
        size_t stub_size;

        int r;

        assert(j);

        if (j->payload_size < MAX(sizeof(xz_signature), sizeof(gzip_signature)))
                return 0;

        if (memcmp(j->payload, xz_signature, sizeof(xz_signature)) == 0)
                j->compressed = IMPORT_JOB_XZ;
        else if (memcmp(j->payload, gzip_signature, sizeof(gzip_signature)) == 0)
                j->compressed = IMPORT_JOB_GZIP;
        else
                j->compressed = IMPORT_JOB_UNCOMPRESSED;

        log_debug("Stream is XZ compressed: %s", yes_no(j->compressed == IMPORT_JOB_XZ));
        log_debug("Stream is GZIP compressed: %s", yes_no(j->compressed == IMPORT_JOB_GZIP));

        if (j->compressed == IMPORT_JOB_XZ) {
                lzma_ret xzr;

                xzr = lzma_stream_decoder(&j->xz, UINT64_MAX, LZMA_TELL_UNSUPPORTED_CHECK);
                if (xzr != LZMA_OK) {
                        log_error("Failed to initialize XZ decoder.");
                        return -EIO;
                }
        }
        if (j->compressed == IMPORT_JOB_GZIP) {
                r = inflateInit2(&j->gzip, 15+16);
                if (r != Z_OK) {
                        log_error("Failed to initialize gzip decoder.");
                        return -EIO;
                }
        }

        r = import_job_open_disk(j);
        if (r < 0)
                return r;

        /* Now, take the payload we read so far, and decompress it */
        stub = j->payload;
        stub_size = j->payload_size;

        j->payload = NULL;
        j->payload_size = 0;

        j->state = IMPORT_JOB_RUNNING;

        r = import_job_write_compressed(j, stub, stub_size);
        if (r < 0)
                return r;

        return 0;
}

static size_t import_job_write_callback(void *contents, size_t size, size_t nmemb, void *userdata) {
        ImportJob *j = userdata;
        size_t sz = size * nmemb;
        int r;

        assert(contents);
        assert(j);

        switch (j->state) {

        case IMPORT_JOB_ANALYZING:
                /* Let's first check what it actually is */

                if (!GREEDY_REALLOC(j->payload, j->payload_allocated, j->payload_size + sz)) {
                        r = log_oom();
                        goto fail;
                }

                memcpy((uint8_t*) j->payload + j->payload_size, contents, sz);
                j->payload_size += sz;

                r = import_job_detect_compression(j);
                if (r < 0)
                        goto fail;

                break;

        case IMPORT_JOB_RUNNING:

                r = import_job_write_compressed(j, contents, sz);
                if (r < 0)
                        goto fail;

                break;

        case IMPORT_JOB_DONE:
        case IMPORT_JOB_FAILED:
                r = -ESTALE;
                goto fail;

        default:
                assert_not_reached("Impossible state.");
        }

        return sz;

fail:
        import_job_finish(j, r);
        return 0;
}

static size_t import_job_header_callback(void *contents, size_t size, size_t nmemb, void *userdata) {
        ImportJob *j = userdata;
        size_t sz = size * nmemb;
        _cleanup_free_ char *length = NULL, *last_modified = NULL;
        char *etag;
        int r;

        assert(contents);
        assert(j);

        if (j->state == IMPORT_JOB_DONE || j->state == IMPORT_JOB_FAILED) {
                r = -ESTALE;
                goto fail;
        }

        assert(j->state == IMPORT_JOB_ANALYZING);

        r = curl_header_strdup(contents, sz, "ETag:", &etag);
        if (r < 0) {
                log_oom();
                goto fail;
        }
        if (r > 0) {
                free(j->etag);
                j->etag = etag;

                if (strv_contains(j->old_etags, j->etag)) {
                        log_info("Image already downloaded. Skipping download.");
                        import_job_finish(j, 0);
                        return sz;
                }

                return sz;
        }

        r = curl_header_strdup(contents, sz, "Content-Length:", &length);
        if (r < 0) {
                log_oom();
                goto fail;
        }
        if (r > 0) {
                (void) safe_atou64(length, &j->content_length);

                if (j->content_length != (uint64_t) -1) {
                        char bytes[FORMAT_BYTES_MAX];

                        if (j->content_length > j->compressed_max) {
                                log_error("Content too large.");
                                r = -EFBIG;
                                goto fail;
                        }

                        log_info("Downloading %s for %s.", format_bytes(bytes, sizeof(bytes), j->content_length), j->url);
                }

                return sz;
        }

        r = curl_header_strdup(contents, sz, "Last-Modified:", &last_modified);
        if (r < 0) {
                log_oom();
                goto fail;
        }
        if (r > 0) {
                (void) curl_parse_http_time(last_modified, &j->mtime);
                return sz;
        }

        return sz;

fail:
        import_job_finish(j, r);
        return 0;
}

static int import_job_progress_callback(void *userdata, curl_off_t dltotal, curl_off_t dlnow, curl_off_t ultotal, curl_off_t ulnow) {
        ImportJob *j = userdata;
        unsigned percent;
        usec_t n;

        assert(j);

        if (dltotal <= 0)
                return 0;

        percent = ((100 * dlnow) / dltotal);
        n = now(CLOCK_MONOTONIC);

        if (n > j->last_status_usec + USEC_PER_SEC &&
            percent != j->progress_percent &&
            dlnow < dltotal) {
                char buf[FORMAT_TIMESPAN_MAX];

                if (n - j->start_usec > USEC_PER_SEC && dlnow > 0) {
                        usec_t left, done;

                        done = n - j->start_usec;
                        left = (usec_t) (((double) done * (double) dltotal) / dlnow) - done;

                        log_info("Got %u%% of %s. %s left.", percent, j->url, format_timespan(buf, sizeof(buf), left, USEC_PER_SEC));
                } else
                        log_info("Got %u%% of %s.", percent, j->url);

                j->progress_percent = percent;
                j->last_status_usec = n;
        }

        return 0;
}

int import_job_new(ImportJob **ret, const char *url, CurlGlue *glue, void *userdata) {
        _cleanup_(import_job_unrefp) ImportJob *j = NULL;

        assert(url);
        assert(glue);
        assert(ret);

        j = new0(ImportJob, 1);
        if (!j)
                return -ENOMEM;

        j->state = IMPORT_JOB_INIT;
        j->disk_fd = -1;
        j->userdata = userdata;
        j->glue = glue;
        j->content_length = (uint64_t) -1;
        j->start_usec = now(CLOCK_MONOTONIC);
        j->compressed_max = j->uncompressed_max = 8LLU * 1024LLU * 1024LLU * 1024LLU; /* 8GB */

        j->url = strdup(url);
        if (!j->url)
                return -ENOMEM;

        *ret = j;
        j = NULL;

        return 0;
}

int import_job_begin(ImportJob *j) {
        int r;

        assert(j);

        if (j->state != IMPORT_JOB_INIT)
                return -EBUSY;

        r = curl_glue_make(&j->curl, j->url, j);
        if (r < 0)
                return r;

        if (!strv_isempty(j->old_etags)) {
                _cleanup_free_ char *cc = NULL, *hdr = NULL;

                cc = strv_join(j->old_etags, ", ");
                if (!cc)
                        return -ENOMEM;

                hdr = strappend("If-None-Match: ", cc);
                if (!hdr)
                        return -ENOMEM;

                j->request_header = curl_slist_new(hdr, NULL);
                if (!j->request_header)
                        return -ENOMEM;

                if (curl_easy_setopt(j->curl, CURLOPT_HTTPHEADER, j->request_header) != CURLE_OK)
                        return -EIO;
        }

        if (curl_easy_setopt(j->curl, CURLOPT_WRITEFUNCTION, import_job_write_callback) != CURLE_OK)
                return -EIO;

        if (curl_easy_setopt(j->curl, CURLOPT_WRITEDATA, j) != CURLE_OK)
                return -EIO;

        if (curl_easy_setopt(j->curl, CURLOPT_HEADERFUNCTION, import_job_header_callback) != CURLE_OK)
                return -EIO;

        if (curl_easy_setopt(j->curl, CURLOPT_HEADERDATA, j) != CURLE_OK)
                return -EIO;

        if (curl_easy_setopt(j->curl, CURLOPT_XFERINFOFUNCTION, import_job_progress_callback) != CURLE_OK)
                return -EIO;

        if (curl_easy_setopt(j->curl, CURLOPT_XFERINFODATA, j) != CURLE_OK)
                return -EIO;

        if (curl_easy_setopt(j->curl, CURLOPT_NOPROGRESS, 0) != CURLE_OK)
                return -EIO;

        r = curl_glue_add(j->glue, j->curl);
        if (r < 0)
                return r;

        j->state = IMPORT_JOB_ANALYZING;

        return 0;
}
Commit	Line	Data
56ebfaf1 LP	1	/-- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil --/
	2
	3	/***
	4	This file is part of systemd.
	5
	6	Copyright 2015 Lennart Poettering
	7
	8	systemd is free software; you can redistribute it and/or modify it
	9	under the terms of the GNU Lesser General Public License as published by
	10	the Free Software Foundation; either version 2.1 of the License, or
	11	(at your option) any later version.
	12
	13	systemd is distributed in the hope that it will be useful, but
	14	WITHOUT ANY WARRANTY; without even the implied warranty of
	15	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	16	Lesser General Public License for more details.
	17
	18	You should have received a copy of the GNU Lesser General Public License
	19	along with systemd; If not, see <http://www.gnu.org/licenses/>.
	20	***/
	21
	22	#include <sys/xattr.h>
	23
	24	#include "strv.h"
	25	#include "import-job.h"
	26
	27	ImportJob* import_job_unref(ImportJob *j) {
	28	if (!j)
	29	return NULL;
	30
	31	curl_glue_remove_and_free(j->glue, j->curl);
	32	curl_slist_free_all(j->request_header);
	33
	34	safe_close(j->disk_fd);
	35
	36	if (j->compressed == IMPORT_JOB_XZ)
	37	lzma_end(&j->xz);
	38	else if (j->compressed == IMPORT_JOB_GZIP)
	39	inflateEnd(&j->gzip);
	40
	41	free(j->url);
	42	free(j->etag);
	43	strv_free(j->old_etags);
	44	free(j->payload);
	45
	46	free(j);
	47
	48	return NULL;
	49	}
	50
	51	DEFINE_TRIVIAL_CLEANUP_FUNC(ImportJob*, import_job_unref);
	52
	53	static void import_job_finish(ImportJob *j, int ret) {
	54	assert(j);
	55
	56	if (j->state == IMPORT_JOB_DONE \|\|
	57	j->state == IMPORT_JOB_FAILED)
	58	return;
	59
68c913fd	60	if (ret == 0) {
56ebfaf1	61	j->state = IMPORT_JOB_DONE;
68c913fd LP	62	log_info("Download of %s complete.", j->url);
68c913fd LP	63	} else {
56ebfaf1 LP	64	j->state = IMPORT_JOB_FAILED;
	65	j->error = ret;
	66	}
	67
	68	if (j->on_finished)
	69	j->on_finished(j);
	70	}
	71
	72	void import_job_curl_on_finished(CurlGlue g, CURL curl, CURLcode result) {
	73	ImportJob *j = NULL;
	74	CURLcode code;
	75	long status;
	76	int r;
	77
	78	if (curl_easy_getinfo(curl, CURLINFO_PRIVATE, &j) != CURLE_OK)
	79	return;
	80
	81	if (!j \|\| j->state == IMPORT_JOB_DONE \|\| j->state == IMPORT_JOB_FAILED)
	82	return;
	83
	84	if (result != CURLE_OK) {
	85	log_error("Transfer failed: %s", curl_easy_strerror(result));
	86	r = -EIO;
	87	goto finish;
	88	}
	89
	90	code = curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &status);
	91	if (code != CURLE_OK) {
	92	log_error("Failed to retrieve response code: %s", curl_easy_strerror(code));
	93	r = -EIO;
	94	goto finish;
	95	} else if (status == 304) {
	96	log_info("Image already downloaded. Skipping download.");
	97	r = 0;
	98	goto finish;
	99	} else if (status >= 300) {
	100	log_error("HTTP request to %s failed with code %li.", j->url, status);
	101	r = -EIO;
	102	goto finish;
	103	} else if (status < 200) {
	104	log_error("HTTP request to %s finished with unexpected code %li.", j->url, status);
	105	r = -EIO;
	106	goto finish;
	107	}
	108
	109	if (j->state != IMPORT_JOB_RUNNING) {
	110	log_error("Premature connection termination.");
	111	r = -EIO;
	112	goto finish;
	113	}
	114
	115	if (j->content_length != (uint64_t) -1 &&
	116	j->content_length != j->written_compressed) {
	117	log_error("Download truncated.");
	118	r = -EIO;
	119	goto finish;
	120	}
	121
	122	if (j->disk_fd >= 0 && j->allow_sparse) {
	123	/* Make sure the file size is right, in case the file was
	124	* sparse and we just seeked for the last part */
	125
	126	if (ftruncate(j->disk_fd, j->written_uncompressed) < 0) {
	127	log_error_errno(errno, "Failed to truncate file: %m");
128	r = -errno;
129	goto finish;
130	}
131
132	if (j->etag)
133	(void) fsetxattr(j->disk_fd, "user.source_etag", j->etag, strlen(j->etag), 0);
134	if (j->url)
135	(void) fsetxattr(j->disk_fd, "user.source_url", j->url, strlen(j->url), 0);
136
137	if (j->mtime != 0) {
138	struct timespec ut[2];
139
140	timespec_store(&ut[0], j->mtime);
141	ut[1] = ut[0];
142	(void) futimens(j->disk_fd, ut);
143
144	(void) fd_setcrtime(j->disk_fd, j->mtime);
145	}
146	}
147
148	r = 0;
149
150	finish:
151	import_job_finish(j, r);
152	}
153
154
155	static int import_job_write_uncompressed(ImportJob j, void p, size_t sz) {
156	ssize_t n;
157
158	assert(j);
159	assert(p);
160	assert(sz > 0);
161	assert(j->disk_fd >= 0);
162
163	if (j->written_uncompressed + sz < j->written_uncompressed) {
164	log_error("File too large, overflow");
165	return -EOVERFLOW;
166	}
167
168	if (j->written_uncompressed + sz > j->uncompressed_max) {
169	log_error("File overly large, refusing");
170	return -EFBIG;
171	}
172
173	if (j->disk_fd >= 0) {
174
175	if (j->allow_sparse)
176	n = sparse_write(j->disk_fd, p, sz, 64);
177	else
178	n = write(j->disk_fd, p, sz);
179	if (n < 0) {
180	log_error_errno(errno, "Failed to write file: %m");
181	return -errno;
182	}
183	if ((size_t) n < sz) {
184	log_error("Short write");
185	return -EIO;
186	}
187	} else {
188
189	if (!GREEDY_REALLOC(j->payload, j->payload_allocated, j->payload_size + sz))
190	return log_oom();
191
192	memcpy((uint8_t*) j->payload + j->payload_size, p, sz);
193	j->payload_size += sz;
194	}
195
196	j->written_uncompressed += sz;
197
198	return 0;
199	}
200
201	static int import_job_write_compressed(ImportJob j, void p, size_t sz) {
202	int r;
203
204	assert(j);
205	assert(p);
206	assert(sz > 0);
207	assert(j->disk_fd >= 0);
208
209	if (j->written_compressed + sz < j->written_compressed) {
210	log_error("File too large, overflow");
211	return -EOVERFLOW;
212	}
213
214	if (j->written_compressed + sz > j->compressed_max) {
215	log_error("File overly large, refusing.");
216	return -EFBIG;
217	}
218
219	if (j->content_length != (uint64_t) -1 &&
220	j->written_compressed + sz > j->content_length) {
221	log_error("Content length incorrect.");
222	return -EFBIG;
223	}
224
225	switch (j->compressed) {
226
227	case IMPORT_JOB_UNCOMPRESSED:
228	r = import_job_write_uncompressed(j, p, sz);
229	if (r < 0)
230	return r;
231
232	break;
233
234	case IMPORT_JOB_XZ:
235	j->xz.next_in = p;
236	j->xz.avail_in = sz;
237
238	while (j->xz.avail_in > 0) {
239	uint8_t buffer[16 * 1024];
240	lzma_ret lzr;
241
242	j->xz.next_out = buffer;
243	j->xz.avail_out = sizeof(buffer);
244
245	lzr = lzma_code(&j->xz, LZMA_RUN);
246	if (lzr != LZMA_OK && lzr != LZMA_STREAM_END) {
247	log_error("Decompression error.");
248	return -EIO;
249	}
250
251	r = import_job_write_uncompressed(j, buffer, sizeof(buffer) - j->xz.avail_out);
252	if (r < 0)
253	return r;
254	}
255
256	break;
257
258	case IMPORT_JOB_GZIP:
259	j->gzip.next_in = p;
260	j->gzip.avail_in = sz;
261
262	while (j->gzip.avail_in > 0) {
263	uint8_t buffer[16 * 1024];
264
265	j->gzip.next_out = buffer;
266	j->gzip.avail_out = sizeof(buffer);
267
268	r = inflate(&j->gzip, Z_NO_FLUSH);
269	if (r != Z_OK && r != Z_STREAM_END) {
270	log_error("Decompression error.");
271	return -EIO;
272	}
273
274	r = import_job_write_uncompressed(j, buffer, sizeof(buffer) - j->gzip.avail_out);
275	if (r < 0)
276	return r;
277	}
278
279	break;
280
281	default:
282	assert_not_reached("Unknown compression");
283	}
284
285	j->written_compressed += sz;
286
287	return 0;
288	}
289
290	static int import_job_open_disk(ImportJob *j) {
291	int r;
292
293	assert(j);
294
295	if (j->on_open_disk) {
296	r = j->on_open_disk(j);
297	if (r < 0)
298	return r;
299	}
300
301	if (j->disk_fd >= 0) {
302	/* Check if we can do sparse files */
303
304	if (lseek(j->disk_fd, SEEK_SET, 0) == 0)
305	j->allow_sparse = true;
306	else {
307	if (errno != ESPIPE)
308	return log_error_errno(errno, "Failed to seek on file descriptor: %m");
309
310	j->allow_sparse = false;
311	}
312	}
313
314	return 0;
315	}
316
317	static int import_job_detect_compression(ImportJob *j) {
318	static const uint8_t xz_signature[] = {
319	0xfd, '7', 'z', 'X', 'Z', 0x00
320	};
321	static const uint8_t gzip_signature[] = {
322	0x1f, 0x8b
323	};
324
325	_cleanup_free_ uint8_t *stub = NULL;
326	size_t stub_size;
327
328	int r;
329
330	assert(j);
331
332	if (j->payload_size < MAX(sizeof(xz_signature), sizeof(gzip_signature)))
333	return 0;
334
335	if (memcmp(j->payload, xz_signature, sizeof(xz_signature)) == 0)
336	j->compressed = IMPORT_JOB_XZ;
337	else if (memcmp(j->payload, gzip_signature, sizeof(gzip_signature)) == 0)
338	j->compressed = IMPORT_JOB_GZIP;
339	else
340	j->compressed = IMPORT_JOB_UNCOMPRESSED;
341
342	log_debug("Stream is XZ compressed: %s", yes_no(j->compressed == IMPORT_JOB_XZ));
343	log_debug("Stream is GZIP compressed: %s", yes_no(j->compressed == IMPORT_JOB_GZIP));
344
345	if (j->compressed == IMPORT_JOB_XZ) {
346	lzma_ret xzr;
347
348	xzr = lzma_stream_decoder(&j->xz, UINT64_MAX, LZMA_TELL_UNSUPPORTED_CHECK);
349	if (xzr != LZMA_OK) {
350	log_error("Failed to initialize XZ decoder.");
351	return -EIO;
352	}
353	}
354	if (j->compressed == IMPORT_JOB_GZIP) {
355	r = inflateInit2(&j->gzip, 15+16);
356	if (r != Z_OK) {
357	log_error("Failed to initialize gzip decoder.");
358	return -EIO;
359	}
360	}
361
362	r = import_job_open_disk(j);
363	if (r < 0)
364	return r;
365
366	/* Now, take the payload we read so far, and decompress it */
367	stub = j->payload;
368	stub_size = j->payload_size;
369
370	j->payload = NULL;
371	j->payload_size = 0;
372
373	j->state = IMPORT_JOB_RUNNING;
374
375	r = import_job_write_compressed(j, stub, stub_size);
376	if (r < 0)
377	return r;
378
379	return 0;
380	}
381
382	static size_t import_job_write_callback(void contents, size_t size, size_t nmemb, void userdata) {
383	ImportJob *j = userdata;
384	size_t sz = size * nmemb;
385	int r;
386
387	assert(contents);
388	assert(j);
389
390	switch (j->state) {
391
392	case IMPORT_JOB_ANALYZING:
393	/* Let's first check what it actually is */
394
395	if (!GREEDY_REALLOC(j->payload, j->payload_allocated, j->payload_size + sz)) {
396	r = log_oom();
397	goto fail;
398	}
399
400	memcpy((uint8_t*) j->payload + j->payload_size, contents, sz);
401	j->payload_size += sz;
402
403	r = import_job_detect_compression(j);
404	if (r < 0)
405	goto fail;
406
407	break;
408
409	case IMPORT_JOB_RUNNING:
410
411	r = import_job_write_compressed(j, contents, sz);
412	if (r < 0)
413	goto fail;
414
415	break;
416
417	case IMPORT_JOB_DONE:
418	case IMPORT_JOB_FAILED:
419	r = -ESTALE;
420	goto fail;
421
422	default:
423	assert_not_reached("Impossible state.");
424	}
425
426	return sz;
427
428	fail:
429	import_job_finish(j, r);
430	return 0;
431	}
432
433	static size_t import_job_header_callback(void contents, size_t size, size_t nmemb, void userdata) {
434	ImportJob *j = userdata;
435	size_t sz = size * nmemb;
436	_cleanup_free_ char length = NULL, last_modified = NULL;
437	char *etag;
438	int r;
439
440	assert(contents);
441	assert(j);
442
443	if (j->state == IMPORT_JOB_DONE \|\| j->state == IMPORT_JOB_FAILED) {
444	r = -ESTALE;
445	goto fail;
446	}
447
448	assert(j->state == IMPORT_JOB_ANALYZING);
449
450	r = curl_header_strdup(contents, sz, "ETag:", &etag);
451	if (r < 0) {
452	log_oom();
453	goto fail;
454	}
455	if (r > 0) {
456	free(j->etag);
457	j->etag = etag;
458
459	if (strv_contains(j->old_etags, j->etag)) {
460	log_info("Image already downloaded. Skipping download.");
461	import_job_finish(j, 0);
462	return sz;
463	}
464
465	return sz;
466	}
467
468	r = curl_header_strdup(contents, sz, "Content-Length:", &length);
469	if (r < 0) {
470	log_oom();
471	goto fail;
472	}
473	if (r > 0) {
474	(void) safe_atou64(length, &j->content_length);
475
476	if (j->content_length != (uint64_t) -1) {
477	char bytes[FORMAT_BYTES_MAX];
478
479	if (j->content_length > j->compressed_max) {
480	log_error("Content too large.");
481	r = -EFBIG;
482	goto fail;
483	}
484
68c913fd	485	log_info("Downloading %s for %s.", format_bytes(bytes, sizeof(bytes), j->content_length), j->url);
56ebfaf1 LP	486	}
	487
	488	return sz;
	489	}
	490
	491	r = curl_header_strdup(contents, sz, "Last-Modified:", &last_modified);
	492	if (r < 0) {
	493	log_oom();
	494	goto fail;
	495	}
	496	if (r > 0) {
	497	(void) curl_parse_http_time(last_modified, &j->mtime);
	498	return sz;
	499	}
	500
	501	return sz;
	502
	503	fail:
	504	import_job_finish(j, r);
	505	return 0;
	506	}
	507
	508	static int import_job_progress_callback(void *userdata, curl_off_t dltotal, curl_off_t dlnow, curl_off_t ultotal, curl_off_t ulnow) {
	509	ImportJob *j = userdata;
	510	unsigned percent;
	511	usec_t n;
	512
	513	assert(j);
	514
	515	if (dltotal <= 0)
	516	return 0;
	517
	518	percent = ((100 * dlnow) / dltotal);
	519	n = now(CLOCK_MONOTONIC);
	520
	521	if (n > j->last_status_usec + USEC_PER_SEC &&
68c913fd LP	522	percent != j->progress_percent &&
68c913fd LP	523	dlnow < dltotal) {
56ebfaf1 LP	524	char buf[FORMAT_TIMESPAN_MAX];
	525
	526	if (n - j->start_usec > USEC_PER_SEC && dlnow > 0) {
	527	usec_t left, done;
	528
	529	done = n - j->start_usec;
	530	left = (usec_t) (((double) done * (double) dltotal) / dlnow) - done;
	531
	532	log_info("Got %u%% of %s. %s left.", percent, j->url, format_timespan(buf, sizeof(buf), left, USEC_PER_SEC));
	533	} else
	534	log_info("Got %u%% of %s.", percent, j->url);
	535
	536	j->progress_percent = percent;
	537	j->last_status_usec = n;
	538	}
	539
	540	return 0;
	541	}
	542
	543	int import_job_new(ImportJob *ret, const char url, CurlGlue glue, void userdata) {
	544	_cleanup_(import_job_unrefp) ImportJob *j = NULL;
	545
	546	assert(url);
	547	assert(glue);
	548	assert(ret);
	549
	550	j = new0(ImportJob, 1);
	551	if (!j)
	552	return -ENOMEM;
	553
	554	j->state = IMPORT_JOB_INIT;
	555	j->disk_fd = -1;
	556	j->userdata = userdata;
	557	j->glue = glue;
	558	j->content_length = (uint64_t) -1;
	559	j->start_usec = now(CLOCK_MONOTONIC);
	560	j->compressed_max = j->uncompressed_max = 8LLU * 1024LLU * 1024LLU * 1024LLU; /* 8GB */
	561
	562	j->url = strdup(url);
	563	if (!j->url)
	564	return -ENOMEM;
	565
	566	*ret = j;
	567	j = NULL;
	568
	569	return 0;
	570	}
	571
	572	int import_job_begin(ImportJob *j) {
	573	int r;
	574
	575	assert(j);
	576
	577	if (j->state != IMPORT_JOB_INIT)
	578	return -EBUSY;
	579
	580	r = curl_glue_make(&j->curl, j->url, j);
	581	if (r < 0)
	582	return r;
	583
	584	if (!strv_isempty(j->old_etags)) {
	585	_cleanup_free_ char cc = NULL, hdr = NULL;
	586
	587	cc = strv_join(j->old_etags, ", ");
588	if (!cc)
589	return -ENOMEM;
590
591	hdr = strappend("If-None-Match: ", cc);
592	if (!hdr)
593	return -ENOMEM;
594
595	j->request_header = curl_slist_new(hdr, NULL);
596	if (!j->request_header)
597	return -ENOMEM;
598
599	if (curl_easy_setopt(j->curl, CURLOPT_HTTPHEADER, j->request_header) != CURLE_OK)
600	return -EIO;
601	}
602
603	if (curl_easy_setopt(j->curl, CURLOPT_WRITEFUNCTION, import_job_write_callback) != CURLE_OK)
604	return -EIO;
605
606	if (curl_easy_setopt(j->curl, CURLOPT_WRITEDATA, j) != CURLE_OK)
607	return -EIO;
608
609	if (curl_easy_setopt(j->curl, CURLOPT_HEADERFUNCTION, import_job_header_callback) != CURLE_OK)
610	return -EIO;
611
612	if (curl_easy_setopt(j->curl, CURLOPT_HEADERDATA, j) != CURLE_OK)
613	return -EIO;
614
615	if (curl_easy_setopt(j->curl, CURLOPT_XFERINFOFUNCTION, import_job_progress_callback) != CURLE_OK)
616	return -EIO;
617
618	if (curl_easy_setopt(j->curl, CURLOPT_XFERINFODATA, j) != CURLE_OK)
619	return -EIO;
620
621	if (curl_easy_setopt(j->curl, CURLOPT_NOPROGRESS, 0) != CURLE_OK)
622	return -EIO;
623
624	r = curl_glue_add(j->glue, j->curl);
625	if (r < 0)
626	return r;
627
628	j->state = IMPORT_JOB_ANALYZING;
629
630	return 0;
631	}