mirror of
https://github.com/servalproject/serval-dna.git
synced 2025-02-21 09:51:50 +00:00
Re-write manifest parsing, improve rhizome_fetch.c logging
Rhizome manifest parser now parses and validates all known fields, informs about unsupported fields, and unpacks fields into relevant struct manifest elements where appropriate. Is also stricter about whitespace. Rhizome fetch code now logs debug messages if DEBUG_RHIZOME_RX bit is on.
This commit is contained in:
Andrew Bettison
parent
025ccad810
commit
49ee4d13f0
@ -151,6 +151,7 @@ extern sqlite3 *rhizome_db;
|
||||
|
||||
int rhizome_opendb();
|
||||
int rhizome_manifest_createid(rhizome_manifest *m);
|
||||
|
||||
int rhizome_strn_is_manifest_id(const char *text);
|
||||
int rhizome_str_is_manifest_id(const char *text);
|
||||
int rhizome_strn_is_bundle_key(const char *text);
|
||||
@ -159,6 +160,9 @@ int rhizome_strn_is_bundle_crypt_key(const char *text);
|
||||
int rhizome_str_is_bundle_crypt_key(const char *text);
|
||||
int rhizome_strn_is_file_hash(const char *text);
|
||||
int rhizome_str_is_file_hash(const char *text);
|
||||
|
||||
#define alloca_tohex_bid(bid) alloca_tohex((bid), RHIZOME_MANIFEST_ID_BYTES)
|
||||
|
||||
int rhizome_write_manifest_file(rhizome_manifest *m, const char *filename);
|
||||
int rhizome_manifest_selfsign(rhizome_manifest *m);
|
||||
int rhizome_drop_stored_file(const char *id,int maximum_priority);
|
||||
|
||||
218
rhizome_bundle.c
218
rhizome_bundle.c
@ -108,74 +108,182 @@ int rhizome_read_manifest_file(rhizome_manifest *m, const char *filename, int bu
|
||||
m->manifest_all_bytes=m->manifest_bytes;
|
||||
|
||||
/* Parse out variables, signature etc */
|
||||
int ofs=0;
|
||||
while((ofs<m->manifest_bytes)&&(m->manifestdata[ofs]))
|
||||
{
|
||||
int i;
|
||||
char line[1024],var[1024],value[1024];
|
||||
while((ofs<m->manifest_bytes)&&
|
||||
(m->manifestdata[ofs]==0x0a||
|
||||
m->manifestdata[ofs]==0x09||
|
||||
m->manifestdata[ofs]==0x20||
|
||||
m->manifestdata[ofs]==0x0d)) ofs++;
|
||||
for(i=0;(i<(m->manifest_bytes-ofs))
|
||||
&&(i<1023)
|
||||
&&(m->manifestdata[ofs+i]!=0x00)
|
||||
&&(m->manifestdata[ofs+i]!=0x0d)
|
||||
&&(m->manifestdata[ofs+i]!=0x0a);i++)
|
||||
line[i]=m->manifestdata[ofs+i];
|
||||
ofs+=i;
|
||||
line[i]=0;
|
||||
/* Ignore blank lines */
|
||||
if (line[0]==0) continue;
|
||||
/* Ignore comment lines */
|
||||
if (line[0] == '#' || line[0] == '!') continue;
|
||||
/* Parse property lines */
|
||||
/* This could be improved to parse Java's Properties.store() output, by handling backlash
|
||||
escapes and continuation lines */
|
||||
if (sscanf(line,"%[^=]=%[^\n\r]", var, value)==2) {
|
||||
if (rhizome_manifest_get(m,var,NULL,0)) {
|
||||
WARNF("Ill formed manifest file, duplicate variable \"%s\"-- keeping first value)", var);
|
||||
m->errors++;
|
||||
} else if (m->var_count<MAX_MANIFEST_VARS) {
|
||||
/*`
|
||||
if (debug & DEBUG_RHIZOME) {
|
||||
char buf[80];
|
||||
DEBUGF("read manifest line: %s=%s", var, catv(value, buf, sizeof buf));
|
||||
}
|
||||
*/
|
||||
m->vars[m->var_count] = strdup(var);
|
||||
m->values[m->var_count] = strdup(value);
|
||||
if (strcasecmp(var,"id") == 0) {
|
||||
str_toupper_inplace(m->values[m->var_count]);
|
||||
/* Parse hex string of ID into public key, and force to upper case. */
|
||||
if (fromhexstr(m->cryptoSignPublic, value, RHIZOME_MANIFEST_ID_BYTES) == -1) {
|
||||
WARNF("Invalid manifest id: %s", value);
|
||||
m->errors++;
|
||||
}
|
||||
} else if (strcasecmp(var,"filehash") == 0 || strcasecmp(var,"BK") == 0) {
|
||||
int have_service = 0;
|
||||
int have_id = 0;
|
||||
int have_version = 0;
|
||||
int have_date = 0;
|
||||
int have_filesize = 0;
|
||||
int have_filehash = 0;
|
||||
int ofs = 0;
|
||||
while (ofs < m->manifest_bytes && m->manifestdata[ofs]) {
|
||||
char line[1024];
|
||||
int limit = ofs + sizeof line - 1;
|
||||
if (limit > m->manifest_bytes)
|
||||
limit = m->manifest_bytes;
|
||||
char *p = line;
|
||||
while (ofs < limit && !(m->manifestdata[ofs] == '\0' || m->manifestdata[ofs] == '\n' || m->manifestdata[ofs] == '\r'))
|
||||
*p++ = m->manifestdata[ofs++];
|
||||
*p = '\0';
|
||||
if (m->manifestdata[ofs] == '\r')
|
||||
++ofs;
|
||||
if (m->manifestdata[ofs] == '\n')
|
||||
++ofs;
|
||||
/* Ignore blank lines */
|
||||
if (line[0] == '\0')
|
||||
continue;
|
||||
/* Ignore comment lines */
|
||||
if (line[0] == '#' || line[0] == '!')
|
||||
continue;
|
||||
/* Parse field=value lines */
|
||||
size_t linelen = p - line;
|
||||
p = strchr(line, '=');
|
||||
if (p == NULL || p == line) {
|
||||
m->errors++;
|
||||
WARNF(bufferP ? "Malformed manifest line in buffer %p: %s"
|
||||
: "Malformed manifest line in file %s: %s",
|
||||
filename, alloca_toprint(80, (unsigned char *)line, linelen));
|
||||
} else {
|
||||
*p++ = '\0';
|
||||
char *var = line;
|
||||
char *value = p;
|
||||
if (rhizome_manifest_get(m, var, NULL, 0)) {
|
||||
WARNF("Ill formed manifest file, duplicate variable \"%s\"", var);
|
||||
m->errors++;
|
||||
} else if (m->var_count >= MAX_MANIFEST_VARS) {
|
||||
WARN("Ill formed manifest file, too many variables");
|
||||
m->errors++;
|
||||
} else {
|
||||
m->vars[m->var_count] = strdup(var);
|
||||
m->values[m->var_count] = strdup(value);
|
||||
if (strcasecmp(var, "id") == 0) {
|
||||
have_id = 1;
|
||||
if (fromhexstr(m->cryptoSignPublic, value, RHIZOME_MANIFEST_ID_BYTES) == -1) {
|
||||
WARNF("Invalid manifest id: %s", value);
|
||||
m->errors++;
|
||||
} else {
|
||||
/* Force to upper case to avoid case sensitive comparison problems later. */
|
||||
str_toupper_inplace(m->values[m->var_count]);
|
||||
}
|
||||
m->var_count++;
|
||||
}
|
||||
} else {
|
||||
if (debug & DEBUG_RHIZOME) {
|
||||
char buf[80];
|
||||
DEBUGF("Skipping malformed line in manifest file %s: %s", bufferP ? "<buffer>" : filename, catv(line, buf, sizeof buf));
|
||||
} else if (strcasecmp(var, "filehash") == 0) {
|
||||
have_filehash = 1;
|
||||
if (!rhizome_str_is_file_hash(value)) {
|
||||
WARNF("Invalid filehash: %s", value);
|
||||
m->errors++;
|
||||
} else {
|
||||
/* Force to upper case to avoid case sensitive comparison problems later. */
|
||||
str_toupper_inplace(m->values[m->var_count]);
|
||||
strcpy(m->fileHexHash, m->values[m->var_count]);
|
||||
m->fileHashedP = 1;
|
||||
}
|
||||
} else if (strcasecmp(var, "BK") == 0) {
|
||||
if (!rhizome_str_is_bundle_key(value)) {
|
||||
WARNF("Invalid BK: %s", value);
|
||||
m->errors++;
|
||||
} else {
|
||||
/* Force to upper case to avoid case sensitive comparison problems later. */
|
||||
str_toupper_inplace(m->values[m->var_count]);
|
||||
}
|
||||
} else if (strcasecmp(var, "filesize") == 0) {
|
||||
have_filesize = 1;
|
||||
char *ep = value;
|
||||
long long filesize = strtoll(value, &ep, 10);
|
||||
if (ep == value || *ep || filesize < 0) {
|
||||
WARNF("Invalid filesize: %s", value);
|
||||
m->errors++;
|
||||
} else {
|
||||
m->fileLength = filesize;
|
||||
}
|
||||
} else if (strcasecmp(var, "service") == 0) {
|
||||
have_service = 1;
|
||||
if ( strcasecmp(value, RHIZOME_SERVICE_FILE) == 0
|
||||
|| strcasecmp(value, RHIZOME_SERVICE_MESHMS) == 0) {
|
||||
} else {
|
||||
INFOF("Unsupported service: %s", value);
|
||||
// This is not an error... older rhizome nodes must carry newer manifests.
|
||||
}
|
||||
} else if (strcasecmp(var, "version") == 0) {
|
||||
have_version = 1;
|
||||
char *ep = value;
|
||||
long long version = strtoll(value, &ep, 10);
|
||||
if (ep == value || *ep || version < 0) {
|
||||
WARNF("Invalid version: %s", value);
|
||||
m->errors++;
|
||||
} else {
|
||||
m->version = version;
|
||||
}
|
||||
} else if (strcasecmp(var, "date") == 0) {
|
||||
have_date = 1;
|
||||
char *ep = value;
|
||||
long long date = strtoll(value, &ep, 10);
|
||||
if (ep == value || *ep || date < 0) {
|
||||
WARNF("Invalid date: %s", value);
|
||||
m->errors++;
|
||||
}
|
||||
// TODO: store date in manifest struct
|
||||
} else if (strcasecmp(var, "sender") == 0 || strcasecmp(var, "recipient") == 0) {
|
||||
if (!str_is_subscriber_id(value)) {
|
||||
WARNF("Invalid %s: %s", var, value);
|
||||
m->errors++;
|
||||
} else {
|
||||
/* Force to upper case to avoid case sensitive comparison problems later. */
|
||||
str_toupper_inplace(m->values[m->var_count]);
|
||||
}
|
||||
} else if (strcasecmp(var, "name") == 0) {
|
||||
if (value[0] == '\0') {
|
||||
WARNF("Empty name", value);
|
||||
m->errors++;
|
||||
}
|
||||
// TODO: complain if service is not MeshMS
|
||||
} else if (strcasecmp(var, "crypt") == 0) {
|
||||
if (!(strcmp(value, "0") == 0 || strcmp(value, "1") == 0)) {
|
||||
WARNF("Invalid crypt: %s", value);
|
||||
m->errors++;
|
||||
} else {
|
||||
m->payloadEncryption = atoi(value);
|
||||
}
|
||||
} else {
|
||||
INFOF("Unsupported field: %s=%s", var, value);
|
||||
// This is not an error... older rhizome nodes must carry newer manifests.
|
||||
}
|
||||
m->var_count++;
|
||||
}
|
||||
}
|
||||
}
|
||||
/* The null byte gets included in the check sum */
|
||||
if (ofs<m->manifest_bytes) ofs++;
|
||||
if (ofs < m->manifest_bytes)
|
||||
++ofs;
|
||||
|
||||
/* Remember where the text ends */
|
||||
int end_of_text=ofs;
|
||||
m->manifest_bytes = end_of_text;
|
||||
|
||||
if (!have_service) {
|
||||
WARNF("Missing service field");
|
||||
m->errors++;
|
||||
}
|
||||
if (!have_id) {
|
||||
WARNF("Missing manifest id field");
|
||||
m->errors++;
|
||||
}
|
||||
if (!have_version) {
|
||||
WARNF("Missing version field");
|
||||
m->errors++;
|
||||
}
|
||||
if (!have_date) {
|
||||
WARNF("Missing date field");
|
||||
m->errors++;
|
||||
}
|
||||
if (!have_filesize) {
|
||||
WARNF("Missing filesize field");
|
||||
m->errors++;
|
||||
}
|
||||
if (!have_filehash && m->fileLength != 0) {
|
||||
WARNF("Missing filehash field");
|
||||
m->errors++;
|
||||
}
|
||||
|
||||
// TODO Determine group membership here.
|
||||
|
||||
m->manifest_bytes=end_of_text;
|
||||
|
||||
RETURN(0);
|
||||
}
|
||||
|
||||
|
||||
353
rhizome_fetch.c
353
rhizome_fetch.c
@ -382,8 +382,7 @@ int rhizome_position_candidate(int position)
|
||||
}
|
||||
|
||||
/* Verifies manifests as late as possible to avoid wasting time. */
|
||||
int rhizome_suggest_queue_manifest_import(rhizome_manifest *m,
|
||||
struct sockaddr_in *peerip)
|
||||
int rhizome_suggest_queue_manifest_import(rhizome_manifest *m, struct sockaddr_in *peerip)
|
||||
{
|
||||
IN();
|
||||
/* must free manifest when done with it */
|
||||
@ -392,29 +391,22 @@ int rhizome_suggest_queue_manifest_import(rhizome_manifest *m,
|
||||
int priority=100; /* normal priority */
|
||||
int i;
|
||||
|
||||
if (1) DEBUGF("Rhizome considering %s (size=%lld, priority=%d)",
|
||||
id,filesize,priority);
|
||||
m->version=rhizome_manifest_get_ll(m,"version");
|
||||
m->version = rhizome_manifest_get_ll(m,"version");
|
||||
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF("Considering manifest import bid=%s version=%lld size=%lld priority=%d:", id, m->version, filesize, priority);
|
||||
|
||||
if (rhizome_manifest_version_cache_lookup(m)) {
|
||||
/* We already have this version or newer */
|
||||
if (1||debug&DEBUG_RHIZOMESYNC) {
|
||||
DEBUGF("manifest id=%s, version=%lld",
|
||||
rhizome_manifest_get(m,"id",NULL,0),
|
||||
rhizome_manifest_get_ll(m,"version"));
|
||||
DEBUG("We already have that manifest or newer.");
|
||||
}
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUG(" already have that version or newer");
|
||||
rhizome_manifest_free(m);
|
||||
RETURN(-1);
|
||||
} else {
|
||||
if (1||debug&DEBUG_RHIZOMESYNC) {
|
||||
long long stored_version;
|
||||
if (sqlite_exec_int64(&stored_version, "select version from manifests where id='%s'",id) > 0)
|
||||
DEBUGF("manifest id=%s, version=%lld is new to us (we only have version %lld).",
|
||||
id,
|
||||
m->version,
|
||||
stored_version);
|
||||
}
|
||||
}
|
||||
|
||||
if (debug & DEBUG_RHIZOME_RX) {
|
||||
long long stored_version;
|
||||
if (sqlite_exec_int64(&stored_version, "select version from manifests where id='%s'",id) > 0)
|
||||
DEBUGF(" is new (have version %lld)", stored_version);
|
||||
}
|
||||
|
||||
/* work out where to put it in the list */
|
||||
@ -537,7 +529,9 @@ void rhizome_enqueue_suggestions(struct sched_ent *alarm)
|
||||
int rhizome_queue_manifest_import(rhizome_manifest *m, struct sockaddr_in *peerip, int *manifest_kept)
|
||||
{
|
||||
*manifest_kept = 0;
|
||||
int i;
|
||||
|
||||
const char *bid = alloca_tohex_bid(m->cryptoSignPublic);
|
||||
long long filesize = rhizome_manifest_get_ll(m, "filesize");
|
||||
|
||||
/* Do the quick rejection tests first, before the more expensive once,
|
||||
like querying the database for manifests.
|
||||
@ -553,40 +547,33 @@ int rhizome_queue_manifest_import(rhizome_manifest *m, struct sockaddr_in *peeri
|
||||
the cache slot number to implicitly store the first bits.
|
||||
*/
|
||||
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF("Fetching manifest bid=%s version=%lld size=%lld:", bid, m->version, filesize);
|
||||
|
||||
if (rhizome_manifest_version_cache_lookup(m)) {
|
||||
/* We already have this version or newer */
|
||||
if (1||debug&DEBUG_RHIZOMESYNC) {
|
||||
DEBUGF("manifest id=%s, version=%lld",
|
||||
rhizome_manifest_get(m,"id",NULL,0),
|
||||
rhizome_manifest_get_ll(m,"version"));
|
||||
DEBUG("We already have that manifest or newer.");
|
||||
}
|
||||
return -1;
|
||||
} else {
|
||||
if (1||debug&DEBUG_RHIZOMESYNC) {
|
||||
DEBUGF("manifest id=%s, version=%lld is new to us.",
|
||||
rhizome_manifest_get(m,"id",NULL,0),
|
||||
rhizome_manifest_get_ll(m,"version"));
|
||||
}
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUG(" already have that version or newer");
|
||||
return 1;
|
||||
}
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF(" is new");
|
||||
|
||||
/* Don't queue if queue slots already full */
|
||||
if (rhizome_file_fetch_queue_count>=MAX_QUEUED_FILES) {
|
||||
if (1||debug&DEBUG_RHIZOME) DEBUG("Already busy fetching files");
|
||||
return -1;
|
||||
if (rhizome_file_fetch_queue_count >= MAX_QUEUED_FILES) {
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUG(" all fetch queue slots full");
|
||||
return 2;
|
||||
}
|
||||
|
||||
/* Don't queue if already queued */
|
||||
char *id=rhizome_manifest_get(m,"id",NULL,0);
|
||||
if (!rhizome_str_is_manifest_id(id))
|
||||
return WHYF("Invalid manifest ID: %s", id);
|
||||
for(i=0;i<rhizome_file_fetch_queue_count;i++) {
|
||||
rhizome_file_fetch_record
|
||||
*q=&file_fetch_queue[i];
|
||||
if (!strcasecmp(id,rhizome_manifest_get(q->manifest,"id",NULL,0))) {
|
||||
if (1||debug&DEBUG_RHIZOMESYNC)
|
||||
DEBUGF("Already have %s in the queue.",id);
|
||||
return -1;
|
||||
}
|
||||
int i;
|
||||
for (i = 0; i < rhizome_file_fetch_queue_count; ++i) {
|
||||
if (memcmp(m->cryptoSignPublic, file_fetch_queue[i].manifest->cryptoSignPublic, RHIZOME_MANIFEST_ID_BYTES) == 0) {
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF(" manifest fetch already queued");
|
||||
return 3;
|
||||
}
|
||||
}
|
||||
|
||||
if (!rhizome_manifest_get(m, "filehash", m->fileHexHash, sizeof m->fileHexHash))
|
||||
@ -596,143 +583,122 @@ int rhizome_queue_manifest_import(rhizome_manifest *m, struct sockaddr_in *peeri
|
||||
str_toupper_inplace(m->fileHexHash);
|
||||
m->fileHashedP = 1;
|
||||
|
||||
long long filesize = rhizome_manifest_get_ll(m, "filesize");
|
||||
if (filesize > 0 && m->fileHexHash[0]) {
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF(" Getting ready to fetch filehash=%s for bid=%s", m->fileHexHash, bid);
|
||||
|
||||
if (debug & DEBUG_RHIZOMESYNC)
|
||||
DEBUGF("Getting ready to fetch file %s for manifest %s", m->fileHexHash, id);
|
||||
|
||||
if (filesize > 0 && m->fileHexHash[0])
|
||||
{
|
||||
long long gotfile = 0;
|
||||
if (sqlite_exec_int64(&gotfile, "SELECT COUNT(*) FROM FILES WHERE ID='%s' and datavalid=1;", m->fileHexHash) != 1)
|
||||
return WHY("select failed");
|
||||
if (gotfile!=1) {
|
||||
/* We need to get the file */
|
||||
/* Discard request if the same manifest is already queued for reception.
|
||||
*/
|
||||
int i,j;
|
||||
for(i=0;i<rhizome_file_fetch_queue_count;i++)
|
||||
{
|
||||
for(j=0;j<crypto_sign_edwards25519sha512batch_PUBLICKEYBYTES;j++)
|
||||
if (m->cryptoSignPublic[j]
|
||||
!=file_fetch_queue[i].manifest->cryptoSignPublic[j]) break;
|
||||
if (j==crypto_sign_edwards25519sha512batch_PUBLICKEYBYTES)
|
||||
{
|
||||
/* We are already fetching this manifest */
|
||||
if (1||debug&DEBUG_RHIZOME) DEBUGF("Already fetching manifest");
|
||||
return -1;
|
||||
}
|
||||
for(j=0;j<=RHIZOME_FILEHASH_STRLEN;j++)
|
||||
if (m->fileHexHash[j]!=file_fetch_queue[i].fileid[j]) break;
|
||||
if (j==RHIZOME_FILEHASH_STRLEN + 1)
|
||||
{
|
||||
/* We are already fetching this file */
|
||||
if (1||debug&DEBUG_RHIZOME) DEBUGF("Already fetching file %s", m->fileHexHash);
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
|
||||
if (peerip)
|
||||
{
|
||||
/* Transfer via HTTP over IPv4 */
|
||||
int sock = socket(AF_INET, SOCK_STREAM, 0);
|
||||
if (sock == -1)
|
||||
return WHY_perror("socket");
|
||||
if (set_nonblock(sock) == -1) {
|
||||
close(sock);
|
||||
return -1;
|
||||
}
|
||||
struct sockaddr_in addr = *peerip;
|
||||
addr.sin_family = AF_INET;
|
||||
INFOF("HTTP CONNECT family=%u port=%u addr=%u.%u.%u.%u",
|
||||
addr.sin_family, ntohs(addr.sin_port),
|
||||
((unsigned char*)&addr.sin_addr.s_addr)[0],
|
||||
((unsigned char*)&addr.sin_addr.s_addr)[1],
|
||||
((unsigned char*)&addr.sin_addr.s_addr)[2],
|
||||
((unsigned char*)&addr.sin_addr.s_addr)[3]
|
||||
);
|
||||
if (connect(sock, (struct sockaddr*)&addr, sizeof addr) == -1) {
|
||||
if (errno == EINPROGRESS) {
|
||||
if (debug & DEBUG_RHIZOMESYNC)
|
||||
DEBUGF("connect() returned EINPROGRESS");
|
||||
} else {
|
||||
WHY_perror("connect");
|
||||
WHY("Failed to open socket to peer's rhizome web server");
|
||||
close(sock);
|
||||
return -1;
|
||||
}
|
||||
}
|
||||
rhizome_file_fetch_record *q=&file_fetch_queue[rhizome_file_fetch_queue_count];
|
||||
q->manifest = m;
|
||||
*manifest_kept = 1;
|
||||
q->alarm.poll.fd=sock;
|
||||
strncpy(q->fileid, m->fileHexHash, RHIZOME_FILEHASH_STRLEN + 1);
|
||||
snprintf(q->request,1024,"GET /rhizome/file/%s HTTP/1.0\r\n\r\n", q->fileid);
|
||||
q->request_len=strlen(q->request);
|
||||
q->request_ofs=0;
|
||||
q->state=RHIZOME_FETCH_CONNECTING;
|
||||
q->file_len=-1;
|
||||
q->file_ofs=0;
|
||||
|
||||
/* XXX Don't forget to implement resume */
|
||||
/* XXX We should stream file straight into the database */
|
||||
const char *id = rhizome_manifest_get(q->manifest, "id", NULL, 0);
|
||||
if (id == NULL) {
|
||||
close(sock);
|
||||
return WHY("Manifest missing ID");
|
||||
}
|
||||
if (create_rhizome_import_dir() == -1)
|
||||
return -1;
|
||||
char filename[1024];
|
||||
if (!FORM_RHIZOME_IMPORT_PATH(filename, "file.%s", id)) {
|
||||
close(sock);
|
||||
return -1;
|
||||
}
|
||||
q->manifest->dataFileName = strdup(filename);
|
||||
q->file=fopen(q->manifest->dataFileName,"w");
|
||||
if (!q->file) {
|
||||
WHY_perror("fopen");
|
||||
if (1||debug&DEBUG_RHIZOME)
|
||||
DEBUGF("Could not open '%s' to write received file.", q->manifest->dataFileName);
|
||||
close(sock);
|
||||
return -1;
|
||||
}
|
||||
|
||||
/* Watch for activity on the socket */
|
||||
q->alarm.function=rhizome_fetch_poll;
|
||||
fetch_stats.name="rhizome_fetch_poll";
|
||||
q->alarm.stats=&fetch_stats;
|
||||
q->alarm.poll.events=POLLIN|POLLOUT;
|
||||
watch(&q->alarm);
|
||||
/* And schedule a timeout alarm */
|
||||
q->alarm.alarm=overlay_gettime_ms() + RHIZOME_IDLE_TIMEOUT;
|
||||
schedule(&q->alarm);
|
||||
|
||||
rhizome_file_fetch_queue_count++;
|
||||
if (1||debug&DEBUG_RHIZOME) DEBUGF("Queued file for fetching into %s (%d in queue)",
|
||||
q->manifest->dataFileName, rhizome_file_fetch_queue_count);
|
||||
return 0;
|
||||
}
|
||||
else
|
||||
{
|
||||
/* Transfer via overlay */
|
||||
return WHY("Rhizome fetching via overlay not implemented");
|
||||
}
|
||||
long long gotfile = 0;
|
||||
if (sqlite_exec_int64(&gotfile, "SELECT COUNT(*) FROM FILES WHERE ID='%s' and datavalid=1;", m->fileHexHash) != 1)
|
||||
return WHY("select failed");
|
||||
if (gotfile == 0) {
|
||||
/* We need to get the file, unless already queued */
|
||||
int i;
|
||||
for (i = 0; i < rhizome_file_fetch_queue_count; ++i) {
|
||||
if (strcasecmp(m->fileHexHash, file_fetch_queue[i].fileid) == 0) {
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF("Payload fetch already queued, slot %d filehash=%s", m->fileHexHash);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
if (debug&DEBUG_RHIZOMESYNC)
|
||||
DEBUGF("We already have the file for this manifest; importing from manifest alone.");
|
||||
if (create_rhizome_import_dir() == -1)
|
||||
|
||||
if (peerip) {
|
||||
/* Transfer via HTTP over IPv4 */
|
||||
int sock = socket(AF_INET, SOCK_STREAM, 0);
|
||||
if (sock == -1)
|
||||
return WHY_perror("socket");
|
||||
if (set_nonblock(sock) == -1) {
|
||||
close(sock);
|
||||
return -1;
|
||||
}
|
||||
struct sockaddr_in addr = *peerip;
|
||||
addr.sin_family = AF_INET;
|
||||
INFOF("HTTP CONNECT family=%u port=%u addr=%u.%u.%u.%u",
|
||||
addr.sin_family, ntohs(addr.sin_port),
|
||||
((unsigned char*)&addr.sin_addr.s_addr)[0],
|
||||
((unsigned char*)&addr.sin_addr.s_addr)[1],
|
||||
((unsigned char*)&addr.sin_addr.s_addr)[2],
|
||||
((unsigned char*)&addr.sin_addr.s_addr)[3]
|
||||
);
|
||||
if (connect(sock, (struct sockaddr*)&addr, sizeof addr) == -1) {
|
||||
if (errno == EINPROGRESS) {
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF("connect() returned EINPROGRESS");
|
||||
} else {
|
||||
WHY_perror("connect");
|
||||
WHY("Failed to open socket to peer's rhizome web server");
|
||||
close(sock);
|
||||
return -1;
|
||||
char filename[1024];
|
||||
if (!FORM_RHIZOME_IMPORT_PATH(filename, "manifest.%s", id))
|
||||
return -1;
|
||||
if (!rhizome_write_manifest_file(m, filename)) {
|
||||
rhizome_bundle_import(m, NULL, id, m->ttl-1);
|
||||
}
|
||||
}
|
||||
rhizome_file_fetch_record *q=&file_fetch_queue[rhizome_file_fetch_queue_count];
|
||||
q->manifest = m;
|
||||
*manifest_kept = 1;
|
||||
q->alarm.poll.fd=sock;
|
||||
strncpy(q->fileid, m->fileHexHash, RHIZOME_FILEHASH_STRLEN + 1);
|
||||
snprintf(q->request,1024,"GET /rhizome/file/%s HTTP/1.0\r\n\r\n", q->fileid);
|
||||
q->request_len=strlen(q->request);
|
||||
q->request_ofs=0;
|
||||
q->state=RHIZOME_FETCH_CONNECTING;
|
||||
q->file_len=-1;
|
||||
q->file_ofs=0;
|
||||
|
||||
/* XXX Don't forget to implement resume */
|
||||
/* XXX We should stream file straight into the database */
|
||||
const char *id = rhizome_manifest_get(q->manifest, "id", NULL, 0);
|
||||
if (id == NULL) {
|
||||
close(sock);
|
||||
return WHY("Manifest missing ID");
|
||||
}
|
||||
if (create_rhizome_import_dir() == -1)
|
||||
return -1;
|
||||
char filename[1024];
|
||||
if (!FORM_RHIZOME_IMPORT_PATH(filename, "file.%s", id)) {
|
||||
close(sock);
|
||||
return -1;
|
||||
}
|
||||
q->manifest->dataFileName = strdup(filename);
|
||||
if ((q->file = fopen(q->manifest->dataFileName, "w")) == NULL) {
|
||||
WHY_perror("fopen");
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF("Could not open '%s' to write received file", q->manifest->dataFileName);
|
||||
close(sock);
|
||||
return -1;
|
||||
}
|
||||
|
||||
/* Watch for activity on the socket */
|
||||
q->alarm.function=rhizome_fetch_poll;
|
||||
fetch_stats.name="rhizome_fetch_poll";
|
||||
q->alarm.stats=&fetch_stats;
|
||||
q->alarm.poll.events=POLLIN|POLLOUT;
|
||||
watch(&q->alarm);
|
||||
/* And schedule a timeout alarm */
|
||||
q->alarm.alarm=overlay_gettime_ms() + RHIZOME_IDLE_TIMEOUT;
|
||||
schedule(&q->alarm);
|
||||
|
||||
rhizome_file_fetch_queue_count++;
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF("Queued file for fetching into %s (%d in queue)",
|
||||
q->manifest->dataFileName, rhizome_file_fetch_queue_count);
|
||||
return 0;
|
||||
} else {
|
||||
/* TODO: fetch via overlay */
|
||||
return WHY("Rhizome fetching via overlay not implemented");
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF("We already have the file for this manifest; importing from manifest alone.");
|
||||
if (create_rhizome_import_dir() == -1)
|
||||
return -1;
|
||||
char filename[1024];
|
||||
if (!FORM_RHIZOME_IMPORT_PATH(filename, "manifest.%s", bid))
|
||||
return -1;
|
||||
if (!rhizome_write_manifest_file(m, filename)) {
|
||||
rhizome_bundle_import(m, NULL, bid, m->ttl-1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
@ -754,9 +720,8 @@ int rhizome_fetch_close(rhizome_file_fetch_record *q){
|
||||
/* Reduce count of open connections */
|
||||
rhizome_file_fetch_queue_count--;
|
||||
|
||||
if (debug&DEBUG_RHIZOME)
|
||||
DEBUGF("Released rhizome fetch slot (%d used)",
|
||||
rhizome_file_fetch_queue_count);
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF("Released rhizome fetch slot (%d used)", rhizome_file_fetch_queue_count);
|
||||
return 0;
|
||||
}
|
||||
|
||||
@ -821,23 +786,29 @@ void rhizome_fetch_poll(struct sched_ent *alarm)
|
||||
bytes=q->file_len-q->file_ofs;
|
||||
if (fwrite(buffer,bytes,1,q->file)!=1)
|
||||
{
|
||||
if (debug&DEBUG_RHIZOME) DEBUGF("Failed writing %d bytes to file. @ offset %d",bytes,q->file_ofs);
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF("Failed to write %d bytes to file @ offset %d", bytes, q->file_ofs);
|
||||
rhizome_fetch_close(q);
|
||||
return;
|
||||
}
|
||||
q->file_ofs+=bytes;
|
||||
|
||||
} else if (bytes==0) {
|
||||
WHY("Got zero bytes, assume socket dead.");
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUG("Got zero bytes, assume socket dead.");
|
||||
rhizome_fetch_close(q);
|
||||
return;
|
||||
}
|
||||
if (q->file_ofs>=q->file_len)
|
||||
{
|
||||
/* got all of file */
|
||||
if (debug&DEBUG_RHIZOME) DEBUGF("Received all of file via rhizome -- now to import it");
|
||||
if (debug & DEBUG_RHIZOME_RX)
|
||||
DEBUGF("Received all of file via rhizome -- now to import it");
|
||||
{
|
||||
fclose(q->file); q->file=NULL;
|
||||
fclose(q->file);
|
||||
q->file = NULL;
|
||||
// TODO: We already have the manifest struct in memory, should import the bundle
|
||||
// directly from that, not by writing it to a file and re-reading it!
|
||||
const char *id = rhizome_manifest_get(q->manifest, "id", NULL, 0);
|
||||
if (id == NULL)
|
||||
{ WHY("Manifest missing ID"); return; }
|
||||
@ -847,10 +818,10 @@ void rhizome_fetch_poll(struct sched_ent *alarm)
|
||||
if (!FORM_RHIZOME_IMPORT_PATH(filename,"manifest.%s", id))
|
||||
return;
|
||||
/* Do really write the manifest unchanged */
|
||||
if (debug&DEBUG_RHIZOME) {
|
||||
if (debug & DEBUG_RHIZOME_RX) {
|
||||
DEBUGF("manifest has %d signatories",q->manifest->sig_count);
|
||||
DEBUGF("manifest id = %s, len=%d",
|
||||
rhizome_manifest_get(q->manifest,"id",NULL,0),
|
||||
DEBUGF("manifest bid=%s len=%d",
|
||||
rhizome_manifest_get(q->manifest, "id", NULL, 0),
|
||||
q->manifest->manifest_bytes);
|
||||
dump("manifest",&q->manifest->manifestdata[0],
|
||||
q->manifest->manifest_all_bytes);
|
||||
@ -862,7 +833,7 @@ void rhizome_fetch_poll(struct sched_ent *alarm)
|
||||
q->manifest->ttl - 1 /* TTL */);
|
||||
}
|
||||
rhizome_manifest_free(q->manifest);
|
||||
q->manifest=NULL;
|
||||
q->manifest = NULL;
|
||||
}
|
||||
rhizome_fetch_close(q);
|
||||
return;
|
||||
|
||||
@ -43,6 +43,7 @@ configure_servald_server() {
|
||||
executeOk_servald config set log.show_time on
|
||||
executeOk_servald config set debug.rhizome on
|
||||
executeOk_servald config set debug.rhizomesync on
|
||||
executeOk_servald config set debug.rhizomerx on
|
||||
executeOk_servald config set mdp.wifi.tick_ms 100
|
||||
executeOk_servald config set mdp.selfannounce.ticks_per_full_address 1
|
||||
executeOk_servald config set rhizome.fetch_interval_ms 100
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user