1 // -*- mode: cpp; mode: fold -*-
3 // $Id: acquire-worker.cc,v 1.34 2001/05/22 04:42:54 jgg Exp $
4 /* ######################################################################
8 The worker process can startup either as a Configuration prober
9 or as a queue runner. As a configuration prober it only reads the
10 configuration message and
12 ##################################################################### */
14 // Include Files /*{{{*/
17 #include <apt-pkg/acquire.h>
18 #include <apt-pkg/acquire-worker.h>
19 #include <apt-pkg/acquire-item.h>
20 #include <apt-pkg/configuration.h>
21 #include <apt-pkg/error.h>
22 #include <apt-pkg/fileutl.h>
23 #include <apt-pkg/strutl.h>
24 #include <apt-pkg/hashes.h>
44 // Worker::Worker - Constructor for Queue startup /*{{{*/
45 pkgAcquire::Worker::Worker(Queue
*Q
, MethodConfig
*Cnf
, pkgAcquireStatus
*log
) :
46 d(NULL
), OwnerQ(Q
), Log(log
), Config(Cnf
), Access(Cnf
->Access
),
47 CurrentItem(nullptr), CurrentSize(0), TotalSize(0)
52 // Worker::Worker - Constructor for method config startup /*{{{*/
53 pkgAcquire::Worker::Worker(MethodConfig
*Cnf
) : Worker(nullptr, Cnf
, nullptr)
57 // Worker::Construct - Constructor helper /*{{{*/
58 // ---------------------------------------------------------------------
60 void pkgAcquire::Worker::Construct()
69 Debug
= _config
->FindB("Debug::pkgAcquire::Worker",false);
72 // Worker::~Worker - Destructor /*{{{*/
73 // ---------------------------------------------------------------------
75 pkgAcquire::Worker::~Worker()
82 /* Closing of stdin is the signal to exit and die when the process
83 indicates it needs cleanup */
84 if (Config
->NeedsCleanup
== false)
86 ExecWait(Process
,Access
.c_str(),true);
90 // Worker::Start - Start the worker process /*{{{*/
91 // ---------------------------------------------------------------------
92 /* This forks the method and inits the communication channel */
93 bool pkgAcquire::Worker::Start()
95 // Get the method path
96 string Method
= _config
->FindDir("Dir::Bin::Methods") + Access
;
97 if (FileExists(Method
) == false)
99 _error
->Error(_("The method driver %s could not be found."),Method
.c_str());
100 if (Access
== "https")
101 _error
->Notice(_("Is the package %s installed?"), "apt-transport-https");
106 clog
<< "Starting method '" << Method
<< '\'' << endl
;
109 int Pipes
[4] = {-1,-1,-1,-1};
110 if (pipe(Pipes
) != 0 || pipe(Pipes
+2) != 0)
112 _error
->Errno("pipe","Failed to create IPC pipe to subprocess");
113 for (int I
= 0; I
!= 4; I
++)
117 for (int I
= 0; I
!= 4; I
++)
118 SetCloseExec(Pipes
[I
],true);
120 // Fork off the process
121 Process
= ExecFork();
125 dup2(Pipes
[1],STDOUT_FILENO
);
126 dup2(Pipes
[2],STDIN_FILENO
);
127 SetCloseExec(STDOUT_FILENO
,false);
128 SetCloseExec(STDIN_FILENO
,false);
129 SetCloseExec(STDERR_FILENO
,false);
132 Args
[0] = Method
.c_str();
134 execv(Args
[0],(char **)Args
);
135 cerr
<< "Failed to exec method " << Args
[0] << endl
;
142 SetNonBlock(Pipes
[0],true);
143 SetNonBlock(Pipes
[3],true);
149 // Read the configuration data
150 if (WaitFd(InFd
) == false ||
151 ReadMessages() == false)
152 return _error
->Error(_("Method %s did not start correctly"),Method
.c_str());
161 // Worker::ReadMessages - Read all pending messages into the list /*{{{*/
162 // ---------------------------------------------------------------------
164 bool pkgAcquire::Worker::ReadMessages()
166 if (::ReadMessages(InFd
,MessageQueue
) == false)
167 return MethodFailure();
171 // Worker::RunMessage - Empty the message queue /*{{{*/
172 // ---------------------------------------------------------------------
173 /* This takes the messages from the message queue and runs them through
174 the parsers in order. */
175 enum class APT_HIDDEN MessageType
{
184 GENERAL_FAILURE
= 401,
187 static bool isDoomedItem(pkgAcquire::Item
const * const Itm
)
189 auto const TransItm
= dynamic_cast<pkgAcqTransactionItem
const * const>(Itm
);
190 if (TransItm
== nullptr)
192 return TransItm
->TransactionManager
->State
!= pkgAcqTransactionItem::TransactionStarted
;
194 bool pkgAcquire::Worker::RunMessages()
196 while (MessageQueue
.empty() == false)
198 string Message
= MessageQueue
.front();
199 MessageQueue
.erase(MessageQueue
.begin());
202 clog
<< " <- " << Access
<< ':' << QuoteString(Message
,"\n") << endl
;
204 // Fetch the message number
206 MessageType
const Number
= static_cast<MessageType
>(strtoul(Message
.c_str(),&End
,10));
207 if (End
== Message
.c_str())
208 return _error
->Error("Invalid message from method %s: %s",Access
.c_str(),Message
.c_str());
210 string URI
= LookupTag(Message
,"URI");
211 pkgAcquire::Queue::QItem
*Itm
= NULL
;
212 if (URI
.empty() == false)
213 Itm
= OwnerQ
->FindItem(URI
,this);
217 // update used mirror
218 string UsedMirror
= LookupTag(Message
,"UsedMirror", "");
219 if (UsedMirror
.empty() == false)
221 for (pkgAcquire::Queue::QItem::owner_iterator O
= Itm
->Owners
.begin(); O
!= Itm
->Owners
.end(); ++O
)
222 (*O
)->UsedMirror
= UsedMirror
;
224 if (Itm
->Description
.find(" ") != string::npos
)
225 Itm
->Description
.replace(0, Itm
->Description
.find(" "), UsedMirror
);
229 // Determine the message number and dispatch
232 case MessageType::CAPABILITIES
:
233 if (Capabilities(Message
) == false)
234 return _error
->Error("Unable to process Capabilities message from %s",Access
.c_str());
237 case MessageType::LOG
:
239 clog
<< " <- (log) " << LookupTag(Message
,"Message") << endl
;
242 case MessageType::STATUS
:
243 Status
= LookupTag(Message
,"Message");
246 case MessageType::REDIRECT
:
250 _error
->Error("Method gave invalid 103 Redirect message");
254 std::string
const NewURI
= LookupTag(Message
,"New-URI",URI
.c_str());
259 // Change the status so that it can be dequeued
260 for (auto const &O
: Itm
->Owners
)
261 O
->Status
= pkgAcquire::Item::StatIdle
;
262 // Mark the item as done (taking care of all queues)
263 // and then put it in the main queue again
264 std::vector
<Item
*> const ItmOwners
= Itm
->Owners
;
265 OwnerQ
->ItemDone(Itm
);
267 for (auto const &Owner
: ItmOwners
)
269 pkgAcquire::ItemDesc
&desc
= Owner
->GetItemDesc();
273 // if we change site, treat it as a mirror change
274 if (URI::SiteOnly(NewURI
) != URI::SiteOnly(desc
.URI
))
276 auto const firstSpace
= desc
.Description
.find(" ");
277 if (firstSpace
!= std::string::npos
)
279 std::string
const OldSite
= desc
.Description
.substr(0, firstSpace
);
280 if (likely(APT::String::Startswith(desc
.URI
, OldSite
)))
282 std::string
const OldExtra
= desc
.URI
.substr(OldSite
.length() + 1);
283 if (likely(APT::String::Endswith(NewURI
, OldExtra
)))
285 std::string
const NewSite
= NewURI
.substr(0, NewURI
.length() - OldExtra
.length());
286 Owner
->UsedMirror
= URI::ArchiveOnly(NewSite
);
287 desc
.Description
.replace(0, firstSpace
, Owner
->UsedMirror
);
293 if (isDoomedItem(Owner
) == false)
294 OwnerQ
->Owner
->Enqueue(desc
);
299 case MessageType::WARNING
:
300 _error
->Warning("%s: %s", Itm
->Owner
->DescURI().c_str(), LookupTag(Message
,"Message").c_str());
303 case MessageType::URI_START
:
307 _error
->Error("Method gave invalid 200 URI Start message");
313 TotalSize
= strtoull(LookupTag(Message
,"Size","0").c_str(), NULL
, 10);
314 ResumePoint
= strtoull(LookupTag(Message
,"Resume-Point","0").c_str(), NULL
, 10);
315 for (auto const Owner
: Itm
->Owners
)
317 Owner
->Start(Message
, TotalSize
);
318 // Display update before completion
321 if (Log
->MorePulses
== true)
322 Log
->Pulse(Owner
->GetOwner());
323 Log
->Fetch(Owner
->GetItemDesc());
330 case MessageType::URI_DONE
:
334 _error
->Error("Method gave invalid 201 URI Done message");
338 PrepareFiles("201::URIDone", Itm
);
340 // Display update before completion
341 if (Log
!= 0 && Log
->MorePulses
== true)
342 for (pkgAcquire::Queue::QItem::owner_iterator O
= Itm
->Owners
.begin(); O
!= Itm
->Owners
.end(); ++O
)
343 Log
->Pulse((*O
)->GetOwner());
345 HashStringList ReceivedHashes
;
347 std::string
const givenfilename
= LookupTag(Message
, "Filename");
348 std::string
const filename
= givenfilename
.empty() ? Itm
->Owner
->DestFile
: givenfilename
;
349 // see if we got hashes to verify
350 for (char const * const * type
= HashString::SupportedHashes(); *type
!= NULL
; ++type
)
352 std::string
const tagname
= std::string(*type
) + "-Hash";
353 std::string
const hashsum
= LookupTag(Message
, tagname
.c_str());
354 if (hashsum
.empty() == false)
355 ReceivedHashes
.push_back(HashString(*type
, hashsum
));
357 // not all methods always sent Hashes our way
358 if (ReceivedHashes
.usable() == false)
360 HashStringList
const ExpectedHashes
= Itm
->GetExpectedHashes();
361 if (ExpectedHashes
.usable() == true && RealFileExists(filename
))
363 Hashes
calc(ExpectedHashes
);
364 FileFd
file(filename
, FileFd::ReadOnly
, FileFd::None
);
366 ReceivedHashes
= calc
.GetHashStringList();
370 // only local files can refer other filenames and counting them as fetched would be unfair
371 if (Log
!= NULL
&& Itm
->Owner
->Complete
== false && Itm
->Owner
->Local
== false && givenfilename
== filename
)
372 Log
->Fetched(ReceivedHashes
.FileSize(),atoi(LookupTag(Message
,"Resume-Point","0").c_str()));
375 std::vector
<Item
*> const ItmOwners
= Itm
->Owners
;
376 OwnerQ
->ItemDone(Itm
);
379 bool const isIMSHit
= StringToBool(LookupTag(Message
,"IMS-Hit"),false) ||
380 StringToBool(LookupTag(Message
,"Alt-IMS-Hit"),false);
381 auto const forcedHash
= _config
->Find("Acquire::ForceHash");
382 for (auto const Owner
: ItmOwners
)
384 HashStringList
const ExpectedHashes
= Owner
->GetExpectedHashes();
385 if(_config
->FindB("Debug::pkgAcquire::Auth", false) == true)
387 std::clog
<< "201 URI Done: " << Owner
->DescURI() << endl
388 << "ReceivedHash:" << endl
;
389 for (HashStringList::const_iterator hs
= ReceivedHashes
.begin(); hs
!= ReceivedHashes
.end(); ++hs
)
390 std::clog
<< "\t- " << hs
->toStr() << std::endl
;
391 std::clog
<< "ExpectedHash:" << endl
;
392 for (HashStringList::const_iterator hs
= ExpectedHashes
.begin(); hs
!= ExpectedHashes
.end(); ++hs
)
393 std::clog
<< "\t- " << hs
->toStr() << std::endl
;
397 // decide if what we got is what we expected
398 bool consideredOkay
= false;
399 if ((forcedHash
.empty() && ExpectedHashes
.empty() == false) ||
400 (forcedHash
.empty() == false && ExpectedHashes
.usable()))
402 if (ReceivedHashes
.empty())
404 /* IMS-Hits can't be checked here as we will have uncompressed file,
405 but the hashes for the compressed file. What we have was good through
406 so all we have to ensure later is that we are not stalled. */
407 consideredOkay
= isIMSHit
;
409 else if (ReceivedHashes
== ExpectedHashes
)
410 consideredOkay
= true;
412 consideredOkay
= false;
416 consideredOkay
= !Owner
->HashesRequired();
418 if (consideredOkay
== true)
419 consideredOkay
= Owner
->VerifyDone(Message
, Config
);
420 else // hashsum mismatch
421 Owner
->Status
= pkgAcquire::Item::StatAuthError
;
424 if (consideredOkay
== true)
426 if (isDoomedItem(Owner
) == false)
427 Owner
->Done(Message
, ReceivedHashes
, Config
);
431 Log
->IMSHit(Owner
->GetItemDesc());
433 Log
->Done(Owner
->GetItemDesc());
438 if (isDoomedItem(Owner
) == false)
440 if (Message
.find("\nFailReason:") == std::string::npos
)
442 if (ReceivedHashes
!= ExpectedHashes
)
443 Message
.append("\nFailReason: HashSumMismatch");
445 Message
.append("\nFailReason: WeakHashSums");
447 Owner
->Failed(Message
,Config
);
450 Log
->Fail(Owner
->GetItemDesc());
457 case MessageType::URI_FAILURE
:
461 std::string
const msg
= LookupTag(Message
,"Message");
462 _error
->Error("Method gave invalid 400 URI Failure message: %s", msg
.c_str());
466 PrepareFiles("400::URIFailure", Itm
);
468 // Display update before completion
469 if (Log
!= nullptr && Log
->MorePulses
== true)
470 for (pkgAcquire::Queue::QItem::owner_iterator O
= Itm
->Owners
.begin(); O
!= Itm
->Owners
.end(); ++O
)
471 Log
->Pulse((*O
)->GetOwner());
473 std::vector
<Item
*> const ItmOwners
= Itm
->Owners
;
474 OwnerQ
->ItemDone(Itm
);
477 bool errTransient
= false, errAuthErr
= false;
479 std::string
const failReason
= LookupTag(Message
, "FailReason");
481 auto const reasons
= { "Timeout", "ConnectionRefused",
482 "ConnectionTimedOut", "ResolveFailure", "TmpResolveFailure" };
483 errTransient
= std::find(std::begin(reasons
), std::end(reasons
), failReason
) != std::end(reasons
);
485 if (errTransient
== false)
487 auto const reasons
= { "HashSumMismatch", "WeakHashSums", "MaximumSizeExceeded" };
488 errAuthErr
= std::find(std::begin(reasons
), std::end(reasons
), failReason
) != std::end(reasons
);
492 for (auto const Owner
: ItmOwners
)
494 if (errAuthErr
&& Owner
->GetExpectedHashes().empty() == false)
495 Owner
->Status
= pkgAcquire::Item::StatAuthError
;
496 else if (errTransient
)
497 Owner
->Status
= pkgAcquire::Item::StatTransientNetworkError
;
499 if (isDoomedItem(Owner
) == false)
500 Owner
->Failed(Message
,Config
);
502 Log
->Fail(Owner
->GetItemDesc());
509 case MessageType::GENERAL_FAILURE
:
510 _error
->Error("Method %s General failure: %s",Access
.c_str(),LookupTag(Message
,"Message").c_str());
513 case MessageType::MEDIA_CHANGE
:
514 MediaChange(Message
);
521 // Worker::Capabilities - 100 Capabilities handler /*{{{*/
522 // ---------------------------------------------------------------------
523 /* This parses the capabilities message and dumps it into the configuration
525 bool pkgAcquire::Worker::Capabilities(string Message
)
530 Config
->Version
= LookupTag(Message
,"Version");
531 Config
->SingleInstance
= StringToBool(LookupTag(Message
,"Single-Instance"),false);
532 Config
->Pipeline
= StringToBool(LookupTag(Message
,"Pipeline"),false);
533 Config
->SendConfig
= StringToBool(LookupTag(Message
,"Send-Config"),false);
534 Config
->LocalOnly
= StringToBool(LookupTag(Message
,"Local-Only"),false);
535 Config
->NeedsCleanup
= StringToBool(LookupTag(Message
,"Needs-Cleanup"),false);
536 Config
->Removable
= StringToBool(LookupTag(Message
,"Removable"),false);
541 clog
<< "Configured access method " << Config
->Access
<< endl
;
542 clog
<< "Version:" << Config
->Version
<<
543 " SingleInstance:" << Config
->SingleInstance
<<
544 " Pipeline:" << Config
->Pipeline
<<
545 " SendConfig:" << Config
->SendConfig
<<
546 " LocalOnly: " << Config
->LocalOnly
<<
547 " NeedsCleanup: " << Config
->NeedsCleanup
<<
548 " Removable: " << Config
->Removable
<< endl
;
554 // Worker::MediaChange - Request a media change /*{{{*/
555 // ---------------------------------------------------------------------
557 bool pkgAcquire::Worker::MediaChange(string Message
)
559 int status_fd
= _config
->FindI("APT::Status-Fd",-1);
562 string Media
= LookupTag(Message
,"Media");
563 string Drive
= LookupTag(Message
,"Drive");
564 ostringstream msg
,status
;
565 ioprintf(msg
,_("Please insert the disc labeled: "
567 "in the drive '%s' and press [Enter]."),
568 Media
.c_str(),Drive
.c_str());
569 status
<< "media-change: " // message
570 << Media
<< ":" // media
571 << Drive
<< ":" // drive
572 << msg
.str() // l10n message
575 std::string
const dlstatus
= status
.str();
576 FileFd::Write(status_fd
, dlstatus
.c_str(), dlstatus
.size());
579 if (Log
== 0 || Log
->MediaChange(LookupTag(Message
,"Media"),
580 LookupTag(Message
,"Drive")) == false)
583 snprintf(S
,sizeof(S
),"603 Media Changed\nFailed: true\n\n");
585 clog
<< " -> " << Access
<< ':' << QuoteString(S
,"\n") << endl
;
592 snprintf(S
,sizeof(S
),"603 Media Changed\n\n");
594 clog
<< " -> " << Access
<< ':' << QuoteString(S
,"\n") << endl
;
600 // Worker::SendConfiguration - Send the config to the method /*{{{*/
601 // ---------------------------------------------------------------------
603 bool pkgAcquire::Worker::SendConfiguration()
605 if (Config
->SendConfig
== false)
611 /* Write out all of the configuration directives by walking the
612 configuration tree */
613 std::ostringstream Message
;
614 Message
<< "601 Configuration\n";
615 _config
->Dump(Message
, NULL
, "Config-Item: %F=%V\n", false);
619 clog
<< " -> " << Access
<< ':' << QuoteString(Message
.str(),"\n") << endl
;
620 OutQueue
+= Message
.str();
626 // Worker::QueueItem - Add an item to the outbound queue /*{{{*/
627 // ---------------------------------------------------------------------
628 /* Send a URI Acquire message to the method */
629 bool pkgAcquire::Worker::QueueItem(pkgAcquire::Queue::QItem
*Item
)
634 HashStringList
const hsl
= Item
->GetExpectedHashes();
636 if (isDoomedItem(Item
->Owner
))
639 if (hsl
.usable() == false && Item
->Owner
->HashesRequired() &&
640 _config
->Exists("Acquire::ForceHash") == false)
642 std::string
const Message
= "400 URI Failure"
643 "\nURI: " + Item
->URI
+
644 "\nFilename: " + Item
->Owner
->DestFile
+
645 "\nFailReason: WeakHashSums";
647 auto const ItmOwners
= Item
->Owners
;
648 for (auto &O
: ItmOwners
)
650 O
->Status
= pkgAcquire::Item::StatAuthError
;
651 O
->Failed(Message
, Config
);
653 Log
->Fail(O
->GetItemDesc());
655 // "queued" successfully, the item just instantly failed
659 string Message
= "600 URI Acquire\n";
660 Message
.reserve(300);
661 Message
+= "URI: " + Item
->URI
;
662 Message
+= "\nFilename: " + Item
->Owner
->DestFile
;
664 for (HashStringList::const_iterator hs
= hsl
.begin(); hs
!= hsl
.end(); ++hs
)
665 Message
+= "\nExpected-" + hs
->HashType() + ": " + hs
->HashValue();
667 if (hsl
.FileSize() == 0)
669 unsigned long long FileSize
= Item
->GetMaximumSize();
673 strprintf(MaximumSize
, "%llu", FileSize
);
674 Message
+= "\nMaximum-Size: " + MaximumSize
;
678 Item
->SyncDestinationFiles();
679 Message
+= Item
->Custom600Headers();
682 if (RealFileExists(Item
->Owner
->DestFile
))
684 std::string
const SandboxUser
= _config
->Find("APT::Sandbox::User");
685 ChangeOwnerAndPermissionOfFile("Item::QueueURI", Item
->Owner
->DestFile
.c_str(),
686 SandboxUser
.c_str(), "root", 0600);
690 clog
<< " -> " << Access
<< ':' << QuoteString(Message
,"\n") << endl
;
697 // Worker::OutFdRead - Out bound FD is ready /*{{{*/
698 // ---------------------------------------------------------------------
700 bool pkgAcquire::Worker::OutFdReady()
705 Res
= write(OutFd
,OutQueue
.c_str(),OutQueue
.length());
707 while (Res
< 0 && errno
== EINTR
);
710 return MethodFailure();
712 OutQueue
.erase(0,Res
);
713 if (OutQueue
.empty() == true)
719 // Worker::InFdRead - In bound FD is ready /*{{{*/
720 // ---------------------------------------------------------------------
722 bool pkgAcquire::Worker::InFdReady()
724 if (ReadMessages() == false)
730 // Worker::MethodFailure - Called when the method fails /*{{{*/
731 // ---------------------------------------------------------------------
732 /* This is called when the method is believed to have failed, probably because
734 bool pkgAcquire::Worker::MethodFailure()
736 _error
->Error("Method %s has died unexpectedly!",Access
.c_str());
738 // do not reap the child here to show meaningfull error to the user
739 ExecWait(Process
,Access
.c_str(),false);
748 MessageQueue
.erase(MessageQueue
.begin(),MessageQueue
.end());
753 // Worker::Pulse - Called periodically /*{{{*/
754 // ---------------------------------------------------------------------
756 void pkgAcquire::Worker::Pulse()
758 if (CurrentItem
== 0)
762 if (stat(CurrentItem
->Owner
->DestFile
.c_str(),&Buf
) != 0)
764 CurrentSize
= Buf
.st_size
;
767 // Worker::ItemDone - Called when the current item is finished /*{{{*/
768 // ---------------------------------------------------------------------
770 void pkgAcquire::Worker::ItemDone()
778 void pkgAcquire::Worker::PrepareFiles(char const * const caller
, pkgAcquire::Queue::QItem
const * const Itm
)/*{{{*/
780 if (RealFileExists(Itm
->Owner
->DestFile
))
782 ChangeOwnerAndPermissionOfFile(caller
, Itm
->Owner
->DestFile
.c_str(), "root", "root", 0644);
783 std::string
const filename
= Itm
->Owner
->DestFile
;
784 for (pkgAcquire::Queue::QItem::owner_iterator O
= Itm
->Owners
.begin(); O
!= Itm
->Owners
.end(); ++O
)
786 pkgAcquire::Item
const * const Owner
= *O
;
787 if (Owner
->DestFile
== filename
|| filename
== "/dev/null")
789 RemoveFile("PrepareFiles", Owner
->DestFile
);
790 if (link(filename
.c_str(), Owner
->DestFile
.c_str()) != 0)
792 // different mounts can't happen for us as we download to lists/ by default,
793 // but if the system is reused by others the locations can potentially be on
794 // different disks, so use symlink as poor-men replacement.
795 // FIXME: Real copying as last fallback, but that is costly, so offload to a method preferable
796 if (symlink(filename
.c_str(), Owner
->DestFile
.c_str()) != 0)
797 _error
->Error("Can't create (sym)link of file %s to %s", filename
.c_str(), Owner
->DestFile
.c_str());
803 for (pkgAcquire::Queue::QItem::owner_iterator O
= Itm
->Owners
.begin(); O
!= Itm
->Owners
.end(); ++O
)
804 RemoveFile("PrepareFiles", (*O
)->DestFile
);