apogee/src/controller/outbox.c

#include "outbox.h"

// Submodules
#include "ffdb/fs_list.h"
#include "collections/array.h"
#include "http/client/client.h"
#include "ap/object.h"
#include "rsa-signature-2017/rsa_signature_2017.h"

// Model
#include "model/server.h"
#include "model/crypto/keys.h"
#include "model/crypto/http_sign.h"
#include "model/account.h"
#include "model/activity.h"
#include "model/outbox_envelope.h"
#include "model/peer.h"
#include "model/fetch.h"

// Submodules
#include "http/url.h"
#include "util/format.h"

// Stdlib
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <ctype.h>
#include <sys/stat.h>
#include <dirent.h>

static bool blacklisted( struct outbox_envelope* env )
{
	FILE* f = NULL;
	bool result = false;
	char* line = NULL;
	size_t len = 0;

	if( !env->shared_inbox ) { goto return_false; }

	f = fopen( "data/host_blacklist.txt", "r" );
	if( !f ) { goto return_false; }

	while( -1 != getline( &line, &len, f ) ) {
		len = strlen(line);
		char* item = line;
		while( isspace( item[len-1] ) ) { item[len-1] = '\0'; len -= 1; }
		while( isspace( *item ) ) { ++item; len -= 1; }

		if( !*item ) { continue; }

		//printf( "Blacklist item: '%s'\n", item );
		if( 0 == strncmp( env->shared_inbox, item, len ) ) { goto return_true; }
	}

	goto return_false;

return_true:
	result = true;
	goto cleanup;
return_false:
	result = false;
	goto cleanup;

cleanup:
	if( f ) { fclose(f); }
	free(line);

	return result;
}

static long submit_activity( const char* postdata, const char* inbox, const char* deliver, struct crypto_keys* keys, struct fetch_data* fd, bool use_tor_proxy )
{
	printf( "Using inbox=%s\n", inbox );
	printf( "Using deliver=%s\n", deliver );
	printf( "Using proxy=%c\n", use_tor_proxy ? 'Y' : 'N' );

	struct http_signature hs;
	memset( &hs, 0, sizeof(hs) );
	hs.input.url = deliver;
	hs.input.method = "post";
	hs.input.post_data = postdata;
	if( !http_signature_make( keys, &hs ) ) {
		printf( "! Failed to make HTTP signature\n" );
		return 600;
	}

	char date_header[512];
	snprintf( date_header, sizeof(date_header), "Date: %s", hs.date );

	char sign_header[512];
	snprintf( sign_header, sizeof(sign_header), "Signature: keyId=\"https://%s/owner/actor#mainKey\",headers=\"(request-target) host date content-length digest\",signature=\"%s\"",
		g_server->domain,
		hs.signature
	);

	char user_agent[512];
	snprintf( user_agent, sizeof(user_agent), "User-Agent: curl (Apogee/0.1; +https://%s/owner/actor)", g_server->domain );

	char digest_header[512];
	snprintf( digest_header, sizeof(digest_header), "Digest: %s", hs.digest );

	char content_length_header[512];
	snprintf( content_length_header, sizeof(content_length_header), "Content-Length: %d", hs.content_length );

	char proxy[512];
	snprintf( proxy,512, "socks5h://localhost:%d", g_server->tor_socks_port );

	printf( "Performing post to %s\n", deliver );
	long status_code = -1;
	const void* request_clearnet[] = {
		HTTP_REQ_URL, deliver,
		HTTP_REQ_HEADER, user_agent,
		HTTP_REQ_HEADER, date_header,
		HTTP_REQ_HEADER, sign_header,
		HTTP_REQ_HEADER, content_length_header,
		HTTP_REQ_HEADER, digest_header,
		HTTP_REQ_HEADER, "Content-Type: application/activity+json",
		HTTP_REQ_POSTDATA, postdata,
		HTTP_REQ_RESULT_STATUS, &status_code,
		HTTP_RES_HEADER_CALLBACK, fetch_handle_header, fd,
		NULL
	};

	const void* request_tor[] = {
		HTTP_REQ_URL, deliver,
		HTTP_REQ_HEADER, user_agent,
		HTTP_REQ_HEADER, date_header,
		HTTP_REQ_HEADER, sign_header,
		HTTP_REQ_HEADER, content_length_header,
		HTTP_REQ_HEADER, digest_header,
		HTTP_REQ_HEADER, "Content-Type: application/activity+json",
		HTTP_REQ_POSTDATA, postdata,
		HTTP_REQ_RESULT_STATUS, &status_code,
		HTTP_RES_HEADER_CALLBACK, fetch_handle_header, fd,
		HTTP_REQ_PROXY, proxy,
		HTTP_REQ_TIMEOUT, (void*)10,
		NULL
	};

	// POST to inbox
	const void** request = request_clearnet;
	if( use_tor_proxy ) {
		request = request_tor;
	}

	printf( "\n\nRequest result:\n" );
	fflush(stdout);
	http_client_do(request);
	printf( "\n\n" );
	fflush(stdout);

	http_signature_free( &hs );

	return status_code;
}
static bool delivery_succeeded( long status_code )
{
	switch( status_code ) {
		case 200:
		case 202:
			return true;
	}
	return false;
}

static bool process_envelope( struct outbox_envelope* env, bool fast_only )
{
	bool result = false;
	char* postdata = NULL;
	struct crypto_keys* keys = NULL;
	FILE* f = NULL;
	struct ap_object* act = NULL;
	struct account* to_account = NULL;
	struct peer* p = NULL;
	char* tor_inbox = NULL;

	if( env->sent ) { return false; }
	if( env->retry_after > time(NULL) ) { return false; }
	if( env->account_id == 0 ) { goto discard; }

	// Get outbox URL
	to_account = account_from_id( env->account_id );

	// Get inbox url
	const char* inbox = to_account->inbox;
	if( env->shared_inbox ) {
		inbox = env->shared_inbox;
	}
	if( !inbox ) {
		printf( "No inbox URL available for %s (id=%d), discarding delivery of activity %d.\n", to_account->display_name, to_account->id, env->activity_id );
		goto discard;
	};

	// Try to get peer information
	char domain[512];
	url_get_domain( inbox, domain,sizeof(domain) );
	p = peer_create_from_domain( domain );

	// Handle tor delivery
	if( p->tor_hidden_service ) {
		char* path = strchr( inbox + 9, '/' );
		tor_inbox = aformat( "http://%s%s", p->tor_hidden_service, path );
		printf( "Using TOR inbox: %s\n", tor_inbox );
	}


	struct fetch_data fd;
	fd.p = p;

	printf( "Processing outbox/%d.json\n", env->id );
	printf( "inbox=%s\n", inbox );
	printf( "account_id=%d\n", env->account_id );
	printf( "activity_id=%d\n", env->activity_id );

	if( p && p->admin_disable_outbox ) {
		printf( "Delivery of activities to %s is administratively disabled\n", inbox );
		goto discard;
	}

	// Check blacklist if at least one attempt to deliver has been made
	if( ( env->retries > 0 ) && blacklisted( env ) ) {
		printf( "Matched host blacklist\n" );
		goto discard;
	}

	// Load crypto keys
	keys = crypto_keys_new();
	if( !crypto_keys_load_private( keys, "data/owner/private.pem" ) ) {
		printf( "Failed to load private key\n" );
		return false;
	}

	// Load target account
	act = activity_from_local_id( env->activity_id );
	if( !act ) {
		printf( "! No activity\n" );
		goto failed;
	}

	// Create signature
	ap_activity_create_rsa_signature_2017( act, keys );

	// Create post data
	size_t size;
	{
		FILE* f2 = open_memstream( &postdata, &size );
		ap_object_write_to_FILE( act, f2 );
		fclose(f2);
	}

	// Force null termination
	postdata = realloc( postdata, size + 1 );
	postdata[size] = '\0';
	//printf( "post: %s\n", postdata );

	long status_code;

	if( !fast_only || ( time(NULL) - p->last_hidden_service_delivery < (3600*24*30) ) ) {
		if( !p->admin_disable_tor ) {
			// Try TOR Hidden Service delivery
			if( tor_inbox ) {
				status_code = submit_activity( postdata, inbox, tor_inbox, keys, &fd, true );
				if( delivery_succeeded(status_code) ) {
					p->last_hidden_service_delivery = time(NULL);
					goto success;
				}
			}

			// Try TOR delivery
			status_code = submit_activity( postdata, inbox, inbox, keys, &fd, true );
			if( delivery_succeeded(status_code) ) {
				p->last_tor_delivery = time(NULL);
				goto success;
			}
		}
	}

	if( !fast_only || ( time(NULL) - p->last_successful_delivery < (3600*24*30 ) ) ) {
		// Try clearnet delivery
		status_code = submit_activity( postdata, inbox, inbox, keys, &fd, false );
		if( delivery_succeeded(status_code) ) {
			p->last_successful_delivery = time(NULL);
			goto success;
		}

		// delivery failed
		printf( "\nServer returned status code %ld\n", status_code );
		if( env->retry_after ) {
			env->retries += 1;
		}
		if( p ) {
			p->last_failed_delivery = time(NULL);

			if( p->last_successful_delivery - p->last_failed_delivery > 60*60*24*7 ) {
				// At least one week since last successful delivery
				printf( "Over a week since last successful delivery, marking for discard.\n" );
				env->retries = g_server->outbox_discard_limit;
			}

			peer_save(p);
		}
		if( env->retries >= g_server->outbox_discard_limit ) {
			// Force discard after 10 delivery attempts
			// TODO: change this to a configuration option
			goto discard;
		}
		env->retry_after = time(NULL) + 60 * ( env->retries + 1 ) * ( env->retries + 1 );
		outbox_envelope_save(env);
	}

	goto failed;

success:
	printf( "Submitted successfully\n" );
	if( p ) {
		p->last_successful_delivery = time(NULL);
		peer_save(p);
	}
	goto discard;
cleanup:
	free(tor_inbox);
	ap_object_free(act);
	account_free(to_account);
	free(postdata);
	peer_free(p);

	crypto_keys_free(keys);
	if( f ) { fclose(f); }

	return result;
failed:
	result = false;
	goto cleanup;
discard:
	printf( "Discarding.\n" );
	env->sent = true;
	outbox_envelope_save(env);
	result = true;
	goto cleanup;
}

static bool process_pending( bool fast_only )
{
	DIR* d = opendir("data/outbox");
	if( !d ) { return false; }

	struct dirent* entry;

	bool result = false;

	while( (entry=readdir(d)) ) {
		char* remain;
		int i = strtol(entry->d_name,&remain,10);
		if( 0 != strcmp( remain, ".json" ) ) { continue; }

		struct outbox_envelope* env = outbox_envelope_from_id( i );
		if( env ) {
			if( env->sent ) {
				outbox_envelope_delete(env);
			} else if( process_envelope(env, fast_only ) ) {
				printf( "Envelope data/outbox/%d.json sent\n", i );
				outbox_envelope_delete(env);
				env = NULL;
			}
		}
		outbox_envelope_free(env);
		fflush(stdout);
	}

	closedir(d);

	return result;
}
bool cleanup_box( const char* box );

extern bool terminate;
void process_outbox()
{
	mkdir( "data/outbox", 0750 );

	while( !terminate ) {
		bool activity = false;
		activity |= process_pending(true);
		activity |= process_pending(false);

		if( !activity ) {
			fflush(stdout);
			sleep(1);
		}
	}
}