bbs/clrghouz
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects 1 Releases Wiki Activity

Some BINKP optimisation, implemented crypt, implemented receiving compressed transfers

Browse Source
This commit is contained in:
Deon George
2023-07-02 23:40:08 +10:00
parent f9f9fb5345
commit 6f298d778f
28 changed files with 1614 additions and 904 deletions
Download Patch File Download Diff File Expand all files Collapse all files

328
app/Classes/File/Receive.php
View File

@@ -5,10 +5,12 @@ namespace App\Classes\File;
use Illuminate\Support\Arr;
use Illuminate\Support\Collection;
use Illuminate\Support\Facades\Log;
use Illuminate\Support\Facades\Storage;
use Symfony\Component\HttpFoundation\File\Exception\FileException;
use App\Classes\File;
use App\Classes\{File,Protocol};
use App\Classes\FTN\{InvalidPacketException,Packet};
use App\Exceptions\FileGrewException;
use App\Jobs\{MessageProcess,TicProcess};
use App\Models\Address;
@@ -23,22 +25,26 @@ final class Receive extends Item
{
private const LOGKEY = 'IR-';
private const compression = [
'BZ2',
'GZ',
];
private Address $ao;
private Collection $list;
private ?Item $receiving;
private mixed $f; // File descriptor
private int $start; // Time we started receiving
private int $file_pos; // Current write pointer
private string $file; // Local filename for file received
/** @var ?string The compression used by the incoming file */
private ?string $comp;
/** @var string|null The compressed data received */
private ?string $comp_data;
public function __construct()
{
// Initialise our variables
$this->list = collect();
$this->receiving = NULL;
$this->file_pos = 0;
$this->f = NULL;
}
public function __get($key)
@@ -71,7 +77,7 @@ final class Receive extends Item
case 'total_recv_bytes':
return $this->list
->filter(function($item) { return ($item->action & self::I_RECV) && $item->received === TRUE; })
->sum(function($item) { return $item->file_size; });
->sum(function($item) { return $item->size; });
default:
throw new \Exception('Unknown key: '.$key);
@@ -85,114 +91,121 @@ final class Receive extends Item
*/
public function close(): void
{
if (! $this->f)
if (! $this->receiving)
throw new \Exception('No file to close');
if ($this->file_pos != $this->receiving->file_size) {
Log::warning(sprintf('%s: - Closing [%s], but missing [%d] bytes',self::LOGKEY,$this->receiving->file_name,$this->receiving->file_size-$this->file_pos));
$this->receiving->incomplete = TRUE;
}
if ($this->f) {
if ($this->file_pos !== $this->receiving->size) {
Log::warning(sprintf('%s:- Closing [%s], but missing [%d] bytes',self::LOGKEY,$this->receiving->name,$this->receiving->size-$this->file_pos));
$this->receiving->incomplete = TRUE;
}
$this->receiving->received = TRUE;
$this->receiving->received = TRUE;
$end = time()-$this->start;
Log::debug(sprintf('%s: - Closing [%s], received in [%d]',self::LOGKEY,$this->receiving->file_name,$end));
$end = time()-$this->start;
Log::debug(sprintf('%s:- Closing [%s], received in [%d]',self::LOGKEY,$this->receiving->name,$end));
fclose($this->f);
$this->file_pos = 0;
$this->f = NULL;
if ($this->comp)
Log::info(sprintf('%s:= Compressed file using [%s] was [%d] bytes (%d). Compression rate [%3.2f%%]',self::LOGKEY,$this->comp,$x=strlen($this->comp_data),$this->receiving->size,$x/$this->receiving->size*100));
// If the packet has been received but not the right size, dont process it any more.
fclose($this->f);
// Set our mtime
touch($this->file,$this->mtime);
$this->file_pos = 0;
$this->f = NULL;
// If we received a packet, we'll dispatch a job to process it
if (! $this->receiving->incomplete)
switch ($this->receiving->file_type) {
case self::IS_ARC:
case self::IS_PKT:
Log::info(sprintf('%s: - Processing mail %s [%s]',self::LOGKEY,$this->receiving->file_type === self::IS_PKT ? 'PACKET' : 'ARCHIVE',$this->file));
// If the packet has been received but not the right size, dont process it any more.
try {
$f = new File($this->file);
$processed = FALSE;
// If we received a packet, we'll dispatch a job to process it
if (! $this->receiving->incomplete)
switch ($this->receiving->type) {
case self::IS_ARC:
case self::IS_PKT:
Log::info(sprintf('%s:- Processing mail %s [%s]',self::LOGKEY,$this->receiving->type === self::IS_PKT ? 'PACKET' : 'ARCHIVE',$this->file));
foreach ($f as $packet) {
$po = Packet::process($packet,Arr::get(stream_get_meta_data($packet),'uri'),$f->itemSize(),$this->ao->system);
try {
$f = new File($this->file);
$processed = FALSE;
// Check the messages are from the uplink
if ($this->ao->system->addresses->search(function($item) use ($po) { return $item->id === $po->fftn_o->id; }) === FALSE) {
Log::error(sprintf('%s: ! Packet [%s] is not from this link? [%d]',self::LOGKEY,$po->fftn_o->ftn,$this->ao->system_id));
foreach ($f as $packet) {
$po = Packet::process($packet,Arr::get(stream_get_meta_data($packet),'uri'),$f->itemSize(),$this->ao->system);
break;
}
// Check the messages are from the uplink
if ($this->ao->system->addresses->search(function($item) use ($po) { return $item->id === $po->fftn_o->id; }) === FALSE) {
Log::error(sprintf('%s:! Packet [%s] is not from this link? [%d]',self::LOGKEY,$po->fftn_o->ftn,$this->ao->system_id));
// Check the packet password
if ($this->ao->session('pktpass') != $po->password) {
Log::error(sprintf('%s: ! Packet from [%s] with password [%s] is invalid.',self::LOGKEY,$this->ao->ftn,$po->password));
// @todo Generate message to system advising invalid password - that message should be sent without a packet password!
break;
}
Log::info(sprintf('%s: - Packet has [%d] messages',self::LOGKEY,$po->count()));
// Queue messages if there are too many in the packet.
if ($queue = ($po->count() > config('app.queue_msgs')))
Log::info(sprintf('%s: - Messages will be sent to the queue for processing',self::LOGKEY));
$count = 0;
foreach ($po as $msg) {
Log::info(sprintf('%s: - Mail from [%s] to [%s]',self::LOGKEY,$msg->fftn,$msg->tftn));
// @todo Quick check that the packet should be processed by us.
// @todo validate that the packet's zone is in the domain.
try {
// Dispatch job.
if ($queue)
MessageProcess::dispatch($msg,$f->pktName());
else
MessageProcess::dispatchSync($msg,$f->pktName());
} catch (\Exception $e) {
Log::error(sprintf('%s:! Got error dispatching message [%s] (%d:%s-%s).',self::LOGKEY,$msg->msgid,$e->getLine(),$e->getFile(),$e->getMessage()));
break;
}
$count++;
// Check the packet password
if ($this->ao->session('pktpass') != $po->password) {
Log::error(sprintf('%s:! Packet from [%s] with password [%s] is invalid.',self::LOGKEY,$this->ao->ftn,$po->password));
// @todo Generate message to system advising invalid password - that message should be sent without a packet password!
break;
}
Log::info(sprintf('%s:- Packet has [%d] messages',self::LOGKEY,$po->count()));
// Queue messages if there are too many in the packet.
if ($queue = ($po->count() > config('app.queue_msgs')))
Log::info(sprintf('%s:- Messages will be sent to the queue for processing',self::LOGKEY));
$count = 0;
foreach ($po as $msg) {
Log::info(sprintf('%s:- Mail from [%s] to [%s]',self::LOGKEY,$msg->fftn,$msg->tftn));
// @todo Quick check that the packet should be processed by us.
// @todo validate that the packet's zone is in the domain.
try {
// Dispatch job.
if ($queue)
MessageProcess::dispatch($msg,$f->pktName());
else
MessageProcess::dispatchSync($msg,$f->pktName());
} catch (\Exception $e) {
Log::error(sprintf('%s:! Got error dispatching message [%s] (%d:%s-%s).',self::LOGKEY,$msg->msgid,$e->getLine(),$e->getFile(),$e->getMessage()));
}
$count++;
}
if ($count === $po->count())
$processed = TRUE;
}
if ($count === $po->count())
$processed = TRUE;
if (! $processed) {
Log::alert(sprintf('%s:- Not deleting packet [%s], it doesnt seem to be processed?',self::LOGKEY,$this->file));
// If we want to keep the packet, we could do that logic here
} elseif (! config('app.packet_keep')) {
Log::debug(sprintf('%s:- Deleting processed packet [%s]',self::LOGKEY,$this->file));
unlink($this->file);
}
} catch (InvalidPacketException $e) {
Log::error(sprintf('%s:- Not deleting packet [%s], as it generated an InvalidPacketException',self::LOGKEY,$this->file),['e'=>$e->getMessage()]);
} catch (\Exception $e) {
Log::error(sprintf('%s:- Not deleting packet [%s], as it generated an uncaught exception',self::LOGKEY,$this->file),['e'=>$e->getMessage()]);
}
if (! $processed) {
Log::alert(sprintf('%s: - Not deleting packet [%s], it doesnt seem to be processed?',self::LOGKEY,$this->file));
break;
// If we want to keep the packet, we could do that logic here
} elseif (! config('app.packet_keep')) {
Log::debug(sprintf('%s: - Deleting processed packet [%s]',self::LOGKEY,$this->file));
unlink($this->file);
}
case self::IS_TIC:
Log::info(sprintf('%s:- Processing TIC file [%s]',self::LOGKEY,$this->file));
} catch (InvalidPacketException $e) {
Log::error(sprintf('%s: - Not deleting packet [%s], as it generated an InvalidPacketException',self::LOGKEY,$this->file),['e'=>$e->getMessage()]);
// Queue the tic to be processed later, in case the referenced file hasnt been received yet
TicProcess::dispatch($this->file);
} catch (\Exception $e) {
Log::error(sprintf('%s: - Not deleting packet [%s], as it generated an uncaught exception',self::LOGKEY,$this->file),['e'=>$e->getMessage()]);
}
break;
break;
case self::IS_TIC:
Log::info(sprintf('%s: - Processing TIC file [%s]',self::LOGKEY,$this->file));
// Queue the tic to be processed later, in case the referenced file hasnt been received yet
TicProcess::dispatch($this->file);
break;
default:
Log::debug(sprintf('%s: - Leaving file [%s] in the inbound dir',self::LOGKEY,$this->file));
}
default:
Log::debug(sprintf('%s:- Leaving file [%s] in the inbound dir',self::LOGKEY,$this->file));
}
}
$this->receiving = NULL;
}
@@ -202,38 +215,66 @@ final class Receive extends Item
*
* @param Address $ao
* @param bool $check
* @return bool
* @param string|null $comp If the incoming file will be compressed
* @return int
* @throws \Exception
* @todo $comp should be parsed, in case it contains other items
*/
public function open(Address $ao,bool $check=FALSE): bool
public function open(Address $ao,bool $check=FALSE,string $comp=NULL): int
{
Log::debug(sprintf('%s:+ open [%d]',self::LOGKEY,$check));
// Check we can open this file
// @todo
// @todo implement return 2 - SKIP file
// @todo implement return 4 - SUSPEND(?) file
if ($check) {
return 0;
}
// @todo Change to use Storage::class()
if (! $this->receiving)
throw new \Exception('No files currently receiving');
$this->comp_data = '';
$this->comp = $comp;
/*
if ($this->receiving->size <= self::MAX_COMPSIZE) {
$this->comp = $comp;
} else {
Log::alert(sprintf('%s:- Compression [%s] disabled for file [%s], its size is too big [%d]',self::LOGKEY,$comp,$this->receiving->name,$this->receiving->size));
}
*/
if ($this->comp && (! in_array($this->comp,self::compression)))
throw new \Exception('Unsupported compression:'.$this->comp);
elseif ($this->comp)
Log::debug(sprintf('%s:- Receiving file with [%s] compression',self::LOGKEY,$this->comp));
$this->ao = $ao;
$this->file_pos = 0;
$this->start = time();
$this->file = sprintf('storage/app/%s/%04X-%s',config('app.fido'),$this->ao->id,$this->receiving->recvas);
$this->file = sprintf('storage/app/%s',$this->local_path($ao));
Log::debug(sprintf('%s: - Opening [%s]',self::LOGKEY,$this->file));
$this->f = fopen($this->file,'wb');
if (! $this->f) {
Log::error(sprintf('%s:! Unable to open file [%s] for writing',self::LOGKEY,$this->receiving->file_name));
return 3; // @todo change to const
if (file_exists($this->file)
&& (Storage::disk('local')->lastModified($this->local_path($ao)) === $this->mtime)
&& (Storage::disk('local')->size($this->local_path($ao)) === $this->size)) {
Log::alert(sprintf('%s:- File already exists - skipping [%s]', self::LOGKEY, $this->file));
return Protocol::FOP_SKIP;
} elseif (file_exists($this->file) && (Storage::disk('local')->size($this->local_path($ao)) > 0)) {
Log::alert(sprintf('%s:- File exists with different details - skipping [%s]',self::LOGKEY,$this->file));
return Protocol::FOP_SUSPEND;
} else {
Log::debug(sprintf('%s:- Opening [%s]',self::LOGKEY,$this->file));
}
Log::info(sprintf('%s:= open - File [%s] opened for writing',self::LOGKEY,$this->receiving->file_name));
return 0; // @todo change to const
// If we are only checking, we'll return (NR mode)
if ($check)
return Protocol::FOP_OK;
$this->f = fopen($this->file,'wb');
if (! $this->f) {
Log::error(sprintf('%s:! Unable to open file [%s] for writing',self::LOGKEY,$this->receiving->name));
return Protocol::FOP_ERROR;
}
Log::info(sprintf('%s:= open - File [%s] opened for writing',self::LOGKEY,$this->receiving->name));
return Protocol::FOP_OK;
}
/**
@@ -255,29 +296,76 @@ final class Receive extends Item
$this->receiving = $o;
}
private function local_path(Address $ao): string
{
return sprintf('%s/%04X-%s',config('app.fido'),$ao->id,$this->receiving->recvas);
}
/**
* Write data to the file we are receiving
*
* @param string $buf
* @return int
* @return bool
* @throws \Exception
*/
public function write(string $buf): int
public function write(string $buf): bool
{
if (! $this->f)
throw new \Exception('No file open for read');
throw new \Exception('No file open for write');
if ($this->file_pos+strlen($buf) > $this->receiving->file_size)
throw new \Exception(sprintf('Too many bytes received [%d] (%d)?',$this->file_pos+strlen($buf),$this->receiving->file_size));
$data = '';
$rc = fwrite($this->f,$buf);
// If we are using compression mode, then we need to buffer the right until we have everything
if ($this->comp) {
$this->comp_data .= $buf;
// See if we can uncompress the data yet
switch ($this->comp) {
case 'BZ2':
if (($data=bzdecompress($this->comp_data,TRUE)) === FALSE)
throw new FileException('BZ2 decompression failed?');
elseif (is_numeric($data))
throw new FileException(sprintf('BZ2 decompression failed with (:%d)?',$data));
break;
case 'GZ':
if (($data=gzdeflate($this->comp_data)) === FALSE)
throw new FileException('BZ2 decompression failed?');
break;
}
// Compressed file grew
if (! strlen($data)) {
if (strlen($this->comp_data) > $this->receiving->size) {
fclose($this->f);
$this->f = NULL;
throw new FileGrewException(sprintf('Error compressed file grew, rejecting [%d] -> [%d]', strlen($this->comp_data), $this->receiving->size));
}
return TRUE;
}
} else {
$data = $buf;
}
if ($this->file_pos+strlen($data) > $this->receiving->size)
throw new \Exception(sprintf('Too many bytes received [%d] (%d)?',$this->file_pos+strlen($buf),$this->receiving->size));
$rc = fwrite($this->f,$data);
if ($rc === FALSE)
throw new FileException('Error while writing to file');
$this->file_pos += $rc;
Log::debug(sprintf('%s:- Write [%d] bytes, file pos now [%d] of [%d] (%d)',self::LOGKEY,$rc,$this->file_pos,$this->receiving->file_size,strlen($buf)));
Log::debug(sprintf('%s:- Write [%d] bytes, file pos now [%d] of [%d] (%d)',self::LOGKEY,$rc,$this->file_pos,$this->receiving->size,strlen($buf)));
return $rc;
if (strlen($this->comp_data) > $this->receiving->size)
Log::alert(sprintf('%s:- Compression grew the file during transfer (%d->%d)',self::LOGKEY,$this->receiving->size,strlen($this->comp_data)));
return TRUE;
}
}