XRootD
Loading...
Searching...
No Matches
XrdClXRootDTransport.cc
Go to the documentation of this file.
1//------------------------------------------------------------------------------
2// Copyright (c) 2011-2014 by European Organization for Nuclear Research (CERN)
3// Author: Lukasz Janyst <ljanyst@cern.ch>
4//------------------------------------------------------------------------------
5// This file is part of the XRootD software suite.
6//
7// XRootD is free software: you can redistribute it and/or modify
8// it under the terms of the GNU Lesser General Public License as published by
9// the Free Software Foundation, either version 3 of the License, or
10// (at your option) any later version.
11//
12// XRootD is distributed in the hope that it will be useful,
13// but WITHOUT ANY WARRANTY; without even the implied warranty of
14// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15// GNU General Public License for more details.
16//
17// You should have received a copy of the GNU Lesser General Public License
18// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
19//
20// In applying this licence, CERN does not waive the privileges and immunities
21// granted to it by virtue of its status as an Intergovernmental Organization
22// or submit itself to any jurisdiction.
23//------------------------------------------------------------------------------
24
27#include "XrdCl/XrdClLog.hh"
28#include "XrdCl/XrdClSocket.hh"
29#include "XrdCl/XrdClMessage.hh"
32#include "XrdCl/XrdClUtils.hh"
34#include "XrdCl/XrdClTls.hh"
35#include "XrdNet/XrdNetAddr.hh"
36#include "XrdNet/XrdNetUtils.hh"
39#include "XrdOuc/XrdOucUtils.hh"
40#include "XrdOuc/XrdOucCRC.hh"
42#include "XrdSys/XrdSysTimer.hh"
47#include "XrdSys/XrdSysE2T.hh"
48#include "XrdCl/XrdClTls.hh"
49#include "XrdCl/XrdClSocket.hh"
51#include "XrdVersion.hh"
52
53#include <arpa/inet.h>
54#include <sys/types.h>
55#include <unistd.h>
56#include <dlfcn.h>
57#include <sstream>
58#include <iomanip>
59#include <set>
60#include <limits>
61
62#include <atomic>
63
65
66namespace XrdCl
67{
69 {
71
72 static void UnloadHandler()
73 {
74 UnloadHandler( "root" );
75 UnloadHandler( "xroot" );
76 }
77
78 static void UnloadHandler( const std::string &trProt )
79 {
81 TransportHandler *trHandler = trManager->GetHandler( trProt );
82 trHandler->WaitBeforeExit();
83 }
84
85 void Register( const std::string &protocol )
86 {
87 XrdSysRWLockHelper scope( lock, false ); // obtain write lock
88 std::pair< std::set<std::string>::iterator, bool > ret = protocols.insert( protocol );
89 // if that's the first time we are using the protocol, the sec lib
90 // was just loaded so now's the time to register the atexit handler
91 if( ret.second )
92 {
93 atexit( UnloadHandler );
94 }
95 }
96
99 std::set<std::string> protocols;
100 };
101
102 //----------------------------------------------------------------------------
104 //----------------------------------------------------------------------------
106 {
107 //--------------------------------------------------------------------------
108 // Define the stream status for the link negotiation purposes
109 //--------------------------------------------------------------------------
122
123 //--------------------------------------------------------------------------
124 // Constructor
125 //--------------------------------------------------------------------------
127 serverFlags( 0 )
128 {
129 }
130
132 uint8_t pathId;
133 uint32_t serverFlags;
134 };
135
136 //----------------------------------------------------------------------------
138 //----------------------------------------------------------------------------
140 {
141 StreamSelector( uint16_t size )
142 {
143 //----------------------------------------------------------------------
144 // Subtract one because we shouldn't take into account the control
145 // stream.
146 //----------------------------------------------------------------------
147 strmqueues.resize( size - 1, 0 );
148 }
149
150 //------------------------------------------------------------------------
151 // @param size : number of streams
152 //------------------------------------------------------------------------
153 void AdjustQueues( uint16_t size )
154 {
155 strmqueues.resize( size - 1, 0);
156 }
157
158 //------------------------------------------------------------------------
159 // @param connected : bitarray stating if given sub-stream is connected
160 //
161 // @return : substream number
162 //------------------------------------------------------------------------
163 uint16_t Select( const std::vector<bool> &connected )
164 {
165 uint16_t ret = 0;
166 size_t minval = std::numeric_limits<size_t>::max();
167
168 for( size_t i = 0; i < connected.size() && i < strmqueues.size(); ++i )
169 {
170 if( !connected[i] ) continue;
171
172 if( strmqueues[i] < minval )
173 {
174 ret = i;
175 minval = strmqueues[i];
176 }
177 }
178
179 ++strmqueues[ret];
180 return ret + 1;
181 }
182
183 //--------------------------------------------------------------------------
184 // Update queue for given substream
185 //--------------------------------------------------------------------------
186 void MsgReceived( uint16_t substrm )
187 {
188 if( substrm > 0 )
189 --strmqueues[substrm - 1];
190 }
191
192 private:
193
194 std::vector<size_t> strmqueues;
195 };
196
198 {
199 BindPrefSelector( std::vector<std::string> && bindprefs ) :
200 bindprefs( std::move( bindprefs ) ), next( 0 )
201 {
202 }
203
204 inline const std::string& Get()
205 {
206 std::string &ret = bindprefs[next];
207 ++next;
208 if( next >= bindprefs.size() )
209 next = 0;
210 return ret;
211 }
212
213 private:
214 std::vector<std::string> bindprefs;
215 size_t next;
216 };
217
218 //----------------------------------------------------------------------------
220 //----------------------------------------------------------------------------
222 {
223 //--------------------------------------------------------------------------
224 // Constructor
225 //--------------------------------------------------------------------------
226 XRootDChannelInfo( const URL &url ):
227 serverFlags(0),
229 firstLogIn(true),
230 authBuffer(0),
231 authProtocol(0),
232 authParams(0),
233 authEnv(0),
234 finstcnt(0),
235 openFiles(0),
236 waitBarrier(0),
237 protection(0),
238 protRespBody(0),
239 protRespSize(0),
240 encrypted(false),
241 istpc(false)
242 {
244 memset( sessionId, 0, 16 );
245 memset( oldSessionId, 0, 16 );
246 }
247
248 //--------------------------------------------------------------------------
249 // Destructor
250 //--------------------------------------------------------------------------
252 {
253 delete [] authBuffer;
254 }
255
256 typedef std::vector<XRootDStreamInfo> StreamInfoVector;
257
258 //--------------------------------------------------------------------------
259 // Data
260 //--------------------------------------------------------------------------
261 uint32_t serverFlags;
263 uint8_t sessionId[16];
264 uint8_t oldSessionId[16];
266 std::shared_ptr<SIDManager> sidManager;
272 std::string streamName;
273 std::string authProtocolName;
274 std::set<uint16_t> sentOpens;
275 std::set<uint16_t> sentCloses;
276 std::atomic<uint32_t> finstcnt; // file instance count
277 uint32_t openFiles;
281 unsigned int protRespSize;
282 std::unique_ptr<StreamSelector> strmSelector;
284 bool istpc;
285 std::unique_ptr<BindPrefSelector> bindSelector;
286 std::string logintoken;
288 };
289
290 //----------------------------------------------------------------------------
291 // Constructor
292 //----------------------------------------------------------------------------
294 pSecUnloadHandler( new PluginUnloadHandler() )
295 {
296 }
297
298 //----------------------------------------------------------------------------
299 // Destructor
300 //----------------------------------------------------------------------------
302 {
303 delete pSecUnloadHandler; pSecUnloadHandler = 0;
304 }
305
306 //----------------------------------------------------------------------------
307 // Read message header from socket
308 //----------------------------------------------------------------------------
310 {
311 //--------------------------------------------------------------------------
312 // A new message - allocate the space needed for the header
313 //--------------------------------------------------------------------------
314 if( message.GetCursor() == 0 && message.GetSize() < 8 )
315 message.Allocate( 8 );
316
317 //--------------------------------------------------------------------------
318 // Read the message header
319 //--------------------------------------------------------------------------
320 if( message.GetCursor() < 8 )
321 {
322 size_t leftToBeRead = 8 - message.GetCursor();
323 while( leftToBeRead )
324 {
325 int bytesRead = 0;
326 XRootDStatus status = socket->Read( message.GetBufferAtCursor(),
327 leftToBeRead, bytesRead );
328 if( !status.IsOK() || status.code == suRetry )
329 return status;
330
331 leftToBeRead -= bytesRead;
332 message.AdvanceCursor( bytesRead );
333 }
334 UnMarshallHeader( message );
335
336 uint32_t bodySize = *(uint32_t*)(message.GetBuffer(4));
337 Log *log = DefaultEnv::GetLog();
338 log->Dump( XRootDTransportMsg, "[msg: %p] Expecting %d bytes of message "
339 "body", (void*)&message, bodySize );
340
341 return XRootDStatus( stOK, suDone );
342 }
344 }
345
346 //----------------------------------------------------------------------------
347 // Read message body from socket
348 //----------------------------------------------------------------------------
350 {
351 //--------------------------------------------------------------------------
352 // Retrieve the body
353 //--------------------------------------------------------------------------
354 size_t leftToBeRead = 0;
355 uint32_t bodySize = 0;
357 bodySize = rsphdr->dlen;
358
359 if( message.GetSize() < bodySize + 8 )
360 message.ReAllocate( bodySize + 8 );
361
362 leftToBeRead = bodySize-(message.GetCursor()-8);
363 while( leftToBeRead )
364 {
365 int bytesRead = 0;
366 XRootDStatus status = socket->Read( message.GetBufferAtCursor(), leftToBeRead, bytesRead );
367
368 if( !status.IsOK() || status.code == suRetry )
369 return status;
370
371 leftToBeRead -= bytesRead;
372 message.AdvanceCursor( bytesRead );
373 }
374
375 return XRootDStatus( stOK, suDone );
376 }
377
378 //----------------------------------------------------------------------------
379 // Read more of the message body from socket
380 //----------------------------------------------------------------------------
382 {
384 if( rsphdr->status != kXR_status )
386
387 //--------------------------------------------------------------------------
388 // In case of non kXR_status responses we read all the response, including
389 // data. For kXR_status responses we first read only the remainder of the
390 // header. The header must then be unmarshalled, and then a second call to
391 // GetMore (repeated for suRetry as needed) will read the data.
392 //--------------------------------------------------------------------------
393
394 uint32_t bodySize = rsphdr->dlen;
395 if( bodySize+8 < sizeof( ServerResponseStatus ) )
397 "kXR_status: invalid message size." );
398
400 bodySize += rspst->bdy.dlen;
401
402 if( message.GetSize() < bodySize + 8 )
403 message.ReAllocate( bodySize + 8 );
404
405 size_t leftToBeRead = bodySize-(message.GetCursor()-8);
406 while( leftToBeRead )
407 {
408 int bytesRead = 0;
409 XRootDStatus status = socket->Read( message.GetBufferAtCursor(), leftToBeRead, bytesRead );
410
411 if( !status.IsOK() || status.code == suRetry )
412 return status;
413
414 leftToBeRead -= bytesRead;
415 message.AdvanceCursor( bytesRead );
416 }
417
418 // Unmarchal to message body
419 Log *log = DefaultEnv::GetLog();
421 if( !st.IsOK() && st.code == errDataError )
422 {
423 log->Error( XRootDTransportMsg, "[msg: %p] %s", (void*)&message,
424 st.GetErrorMessage().c_str() );
425 return st;
426 }
427
428 if( !st.IsOK() )
429 {
430 log->Error( XRootDTransportMsg, "[msg: %p] Failed to unmarshall status body.",
431 (void*)&message );
432 return st;
433 }
434
435 return XRootDStatus( stOK, suDone );
436 }
437
438 //----------------------------------------------------------------------------
439 // Initialize channel
440 //----------------------------------------------------------------------------
442 AnyObject &channelData )
443 {
444 XRootDChannelInfo *info = new XRootDChannelInfo( url );
445 XrdSysMutexHelper scopedLock( info->mutex );
446 channelData.Set( info );
447
448 Env *env = DefaultEnv::GetEnv();
449 int streams = DefaultSubStreamsPerChannel;
450 env->GetInt( "SubStreamsPerChannel", streams );
451 if( streams < 1 ) streams = 1;
452 info->stream.resize( streams );
453 info->strmSelector.reset( new StreamSelector( streams ) );
454 info->encrypted = url.IsSecure();
455 info->istpc = url.IsTPC();
456 info->logintoken = url.GetLoginToken();
457 }
458
459 //----------------------------------------------------------------------------
460 // Finalize channel
461 //----------------------------------------------------------------------------
465
466 //----------------------------------------------------------------------------
467 // HandShake
468 //----------------------------------------------------------------------------
470 AnyObject &channelData )
471 {
472 XRootDChannelInfo *info = 0;
473 channelData.Get( info );
474
475 if (!info)
477
478 XrdSysMutexHelper scopedLock( info->mutex );
479
480 if( info->stream.size() <= handShakeData->subStreamId )
481 {
482 Log *log = DefaultEnv::GetLog();
484 "[%s] Internal error: not enough substreams",
485 handShakeData->streamName.c_str() );
487 }
488
489 if( handShakeData->subStreamId == 0 )
490 {
491 info->streamName = handShakeData->streamName;
492 return HandShakeMain( handShakeData, channelData );
493 }
494 return HandShakeParallel( handShakeData, channelData );
495 }
496
497 //----------------------------------------------------------------------------
498 // Hand shake the main stream
499 //----------------------------------------------------------------------------
500 XRootDStatus XRootDTransport::HandShakeMain( HandShakeData *handShakeData,
501 AnyObject &channelData )
502 {
503 XRootDChannelInfo *info = 0;
504 channelData.Get( info );
505
506 if (!info) {
508 "[%s] Internal error: no channel info",
509 handShakeData->streamName.c_str());
511 }
512
513 XRootDStreamInfo &sInfo = info->stream[handShakeData->subStreamId];
514
515 //--------------------------------------------------------------------------
516 // First step - we need to create and initial handshake and send it out
517 //--------------------------------------------------------------------------
518 if( sInfo.status == XRootDStreamInfo::Disconnected ||
519 sInfo.status == XRootDStreamInfo::Broken )
520 {
521 handShakeData->out = GenerateInitialHSProtocol( handShakeData, info,
523 sInfo.status = XRootDStreamInfo::HandShakeSent;
524 return XRootDStatus( stOK, suContinue );
525 }
526
527 //--------------------------------------------------------------------------
528 // Second step - we got the reply message to the initial handshake
529 //--------------------------------------------------------------------------
530 if( sInfo.status == XRootDStreamInfo::HandShakeSent )
531 {
532 XRootDStatus st = ProcessServerHS( handShakeData, info );
533 if( st.IsOK() )
535 else
536 sInfo.status = XRootDStreamInfo::Broken;
537 return st;
538 }
539
540 //--------------------------------------------------------------------------
541 // Third step - we got the response to the protocol request, we need
542 // to process it and send out a login request
543 //--------------------------------------------------------------------------
544 if( sInfo.status == XRootDStreamInfo::HandShakeReceived )
545 {
546 XRootDStatus st = ProcessProtocolResp( handShakeData, info );
547
548 if( !st.IsOK() )
549 {
550 sInfo.status = XRootDStreamInfo::Broken;
551 return st;
552 }
553
554 if( st.code == suRetry )
555 {
556 handShakeData->out = GenerateProtocol( handShakeData, info,
559 return XRootDStatus( stOK, suRetry );
560 }
561
562 handShakeData->out = GenerateLogIn( handShakeData, info );
563 sInfo.status = XRootDStreamInfo::LoginSent;
564 return XRootDStatus( stOK, suContinue );
565 }
566
567 //--------------------------------------------------------------------------
568 // Fourth step - handle the log in response and proceed with the
569 // authentication if required by the server
570 //--------------------------------------------------------------------------
571 if( sInfo.status == XRootDStreamInfo::LoginSent )
572 {
573 XRootDStatus st = ProcessLogInResp( handShakeData, info );
574
575 if( !st.IsOK() )
576 {
577 sInfo.status = XRootDStreamInfo::Broken;
578 return st;
579 }
580
581 if( st.IsOK() && st.code == suDone )
582 {
583 //----------------------------------------------------------------------
584 // If it's not our first log in we need to end the previous session
585 // to make sure that the server noticed our disconnection and closed
586 // all the writable handles that we owned
587 //----------------------------------------------------------------------
588 if( !info->firstLogIn )
589 {
590 handShakeData->out = GenerateEndSession( handShakeData, info );
592 return XRootDStatus( stOK, suContinue );
593 }
594
595 sInfo.status = XRootDStreamInfo::Connected;
596 info->firstLogIn = false;
597 return st;
598 }
599
600 st = DoAuthentication( handShakeData, info );
601 if( !st.IsOK() )
602 sInfo.status = XRootDStreamInfo::Broken;
603 else
604 sInfo.status = XRootDStreamInfo::AuthSent;
605 return st;
606 }
607
608 //--------------------------------------------------------------------------
609 // Fifth step and later - proceed with the authentication
610 //--------------------------------------------------------------------------
611 if( sInfo.status == XRootDStreamInfo::AuthSent )
612 {
613 XRootDStatus st = DoAuthentication( handShakeData, info );
614
615 if( !st.IsOK() )
616 {
617 sInfo.status = XRootDStreamInfo::Broken;
618 return st;
619 }
620
621 if( st.IsOK() && st.code == suDone )
622 {
623 //----------------------------------------------------------------------
624 // If it's not our first log in we need to end the previous session
625 //----------------------------------------------------------------------
626 if( !info->firstLogIn )
627 {
628 handShakeData->out = GenerateEndSession( handShakeData, info );
630 return XRootDStatus( stOK, suContinue );
631 }
632
633 sInfo.status = XRootDStreamInfo::Connected;
634 info->firstLogIn = false;
635 return st;
636 }
637
638 return st;
639 }
640
641 //--------------------------------------------------------------------------
642 // The last step - kXR_endsess returned
643 //--------------------------------------------------------------------------
644 if( sInfo.status == XRootDStreamInfo::EndSessionSent )
645 {
646 XRootDStatus st = ProcessEndSessionResp( handShakeData, info );
647
648 if( st.IsOK() && st.code == suDone )
649 {
650 sInfo.status = XRootDStreamInfo::Connected;
651 }
652 else if( !st.IsOK() )
653 {
654 sInfo.status = XRootDStreamInfo::Broken;
655 }
656
657 return st;
658 }
659
660 return XRootDStatus( stOK, suDone );
661 }
662
663 //----------------------------------------------------------------------------
664 // Hand shake parallel stream
665 //----------------------------------------------------------------------------
666 XRootDStatus XRootDTransport::HandShakeParallel( HandShakeData *handShakeData,
667 AnyObject &channelData )
668 {
669 XRootDChannelInfo *info = 0;
670 channelData.Get( info );
671
672 if (!info) {
674 "[%s] Internal error: no channel info",
675 handShakeData->streamName.c_str());
676 return XRootDStatus(stFatal, errInternal);
677 }
678
679 XRootDStreamInfo &sInfo = info->stream[handShakeData->subStreamId];
680
681 //--------------------------------------------------------------------------
682 // First step - we need to create and initial handshake and send it out
683 //--------------------------------------------------------------------------
684 if( sInfo.status == XRootDStreamInfo::Disconnected ||
685 sInfo.status == XRootDStreamInfo::Broken )
686 {
687 handShakeData->out = GenerateInitialHSProtocol( handShakeData, info,
689 sInfo.status = XRootDStreamInfo::HandShakeSent;
690 return XRootDStatus( stOK, suContinue );
691 }
692
693 //--------------------------------------------------------------------------
694 // Second step - we got the reply message to the initial handshake,
695 // if successful we need to send bind
696 //--------------------------------------------------------------------------
697 if( sInfo.status == XRootDStreamInfo::HandShakeSent )
698 {
699 XRootDStatus st = ProcessServerHS( handShakeData, info );
700 if( st.IsOK() )
702 else
703 sInfo.status = XRootDStreamInfo::Broken;
704 return st;
705 }
706
707 //--------------------------------------------------------------------------
708 // Second step bis - we got the response to the protocol request, we need
709 // to process it and send out a bind request
710 //--------------------------------------------------------------------------
711 if( sInfo.status == XRootDStreamInfo::HandShakeReceived )
712 {
713 XRootDStatus st = ProcessProtocolResp( handShakeData, info );
714
715 if( !st.IsOK() )
716 {
717 sInfo.status = XRootDStreamInfo::Broken;
718 return st;
719 }
720
721 handShakeData->out = GenerateBind( handShakeData, info );
722 sInfo.status = XRootDStreamInfo::BindSent;
723 return XRootDStatus( stOK, suContinue );
724 }
725
726 //--------------------------------------------------------------------------
727 // Third step - we got the response to the kXR_bind
728 //--------------------------------------------------------------------------
729 if( sInfo.status == XRootDStreamInfo::BindSent )
730 {
731 XRootDStatus st = ProcessBindResp( handShakeData, info );
732
733 if( !st.IsOK() )
734 {
735 sInfo.status = XRootDStreamInfo::Broken;
736 return st;
737 }
738 sInfo.status = XRootDStreamInfo::Connected;
739 return XRootDStatus();
740 }
741 return XRootDStatus();
742 }
743
744 //------------------------------------------------------------------------
745 // @return true if handshake has been done and stream is connected,
746 // false otherwise
747 //------------------------------------------------------------------------
749 AnyObject &channelData )
750 {
751 XRootDChannelInfo *info = 0;
752 channelData.Get( info );
753
754 if (!info) {
756 "[%s] Internal error: no channel info",
757 handShakeData->streamName.c_str());
758 return false;
759 }
760
761 XRootDStreamInfo &sInfo = info->stream[handShakeData->subStreamId];
762 return ( sInfo.status == XRootDStreamInfo::Connected );
763 }
764
765 //----------------------------------------------------------------------------
766 // Check if the stream should be disconnected
767 //----------------------------------------------------------------------------
768 bool XRootDTransport::IsStreamTTLElapsed( time_t inactiveTime,
769 AnyObject &channelData )
770 {
771 XRootDChannelInfo *info = 0;
772 channelData.Get( info );
773
774 Env *env = DefaultEnv::GetEnv();
775 Log *log = DefaultEnv::GetLog();
776
777 if (!info) {
779 "Internal error: no channel info, behaving as if TTL has elapsed");
780 return true;
781 }
782
783 //--------------------------------------------------------------------------
784 // Check the TTL settings for the current server
785 //--------------------------------------------------------------------------
786 int ttl;
787 if( info->serverFlags & kXR_isServer )
788 {
790 env->GetInt( "DataServerTTL", ttl );
791 }
792 else
793 {
795 env->GetInt( "LoadBalancerTTL", ttl );
796 }
797
798 //--------------------------------------------------------------------------
799 // See whether we can give a go-ahead for the disconnection
800 //--------------------------------------------------------------------------
801 XrdSysMutexHelper scopedLock( info->mutex );
802 uint16_t allocatedSIDs = info->sidManager->GetNumberOfAllocatedSIDs();
803 log->Dump( XRootDTransportMsg, "[%s] Stream inactive since %lld seconds, "
804 "TTL: %d, allocated SIDs: %d, open files: %d, bound file objects: %d",
805 info->streamName.c_str(), (long long) inactiveTime, ttl, allocatedSIDs,
806 info->openFiles, info->finstcnt.load( std::memory_order_relaxed ) );
807
808 if( info->openFiles != 0 && info->finstcnt.load( std::memory_order_relaxed ) != 0 )
809 return false;
810
811 if( !allocatedSIDs && inactiveTime > ttl )
812 return true;
813
814 return false;
815 }
816
817 //----------------------------------------------------------------------------
818 // Check the stream is broken - ie. TCP connection got broken and
819 // went undetected by the TCP stack
820 //----------------------------------------------------------------------------
822 AnyObject &channelData )
823 {
824 XRootDChannelInfo *info = 0;
825 channelData.Get( info );
826 Env *env = DefaultEnv::GetEnv();
827 Log *log = DefaultEnv::GetLog();
828
829 if (!info) {
831 "Internal error: no channel info, behaving as if stream is broken");
832 return true;
833 }
834
835 int streamTimeout = DefaultStreamTimeout;
836 env->GetInt( "StreamTimeout", streamTimeout );
837
838 XrdSysMutexHelper scopedLock( info->mutex );
839
840 const time_t now = time(0);
841 const bool anySID =
842 info->sidManager->IsAnySIDOldAs( now - streamTimeout );
843
844 log->Dump( XRootDTransportMsg, "[%s] Stream inactive since %lld seconds, "
845 "stream timeout: %d, any SID: %d, wait barrier: %s",
846 info->streamName.c_str(), (long long) inactiveTime, streamTimeout,
847 anySID, Utils::TimeToString(info->waitBarrier).c_str() );
848
849 if( inactiveTime < streamTimeout )
850 return Status();
851
852 if( now < info->waitBarrier )
853 return Status();
854
855 if( !anySID )
856 return Status();
857
859 }
860
861 //----------------------------------------------------------------------------
862 // Multiplex
863 //----------------------------------------------------------------------------
865 {
866 return PathID( 0, 0 );
867 }
868
869 //----------------------------------------------------------------------------
870 // Multiplex
871 //----------------------------------------------------------------------------
873 AnyObject &channelData,
874 PathID *hint )
875 {
876 XRootDChannelInfo *info = 0;
877 channelData.Get( info );
878
879 if (!info) {
881 "Internal error: no channel info, cannot multiplex");
882 return PathID(0,0);
883 }
884
885 XrdSysMutexHelper scopedLock( info->mutex );
886
887 //--------------------------------------------------------------------------
888 // If we're not connected to a data server or we don't know that yet
889 // we stream through 0
890 //--------------------------------------------------------------------------
891 if( !(info->serverFlags & kXR_isServer) || info->stream.size() == 0 )
892 return PathID( 0, 0 );
893
894 //--------------------------------------------------------------------------
895 // Select the streams
896 //--------------------------------------------------------------------------
897 Log *log = DefaultEnv::GetLog();
898 uint16_t upStream = 0;
899 uint16_t downStream = 0;
900
901 if( hint )
902 {
903 upStream = hint->up;
904 downStream = hint->down;
905 }
906 else
907 {
908 upStream = 0;
909 std::vector<bool> connected;
910 connected.reserve( info->stream.size() - 1 );
911 size_t nbConnected = 0;
912 for( size_t i = 1; i < info->stream.size(); ++i )
913 if( info->stream[i].status == XRootDStreamInfo::Connected )
914 {
915 connected.push_back( true );
916 ++nbConnected;
917 }
918 else
919 connected.push_back( false );
920
921 if( nbConnected == 0 )
922 downStream = 0;
923 else
924 downStream = info->strmSelector->Select( connected );
925 }
926
927 if( upStream >= info->stream.size() )
928 {
930 "[%s] Up link stream %d does not exist, using 0",
931 info->streamName.c_str(), upStream );
932 upStream = 0;
933 }
934
935 if( downStream >= info->stream.size() )
936 {
938 "[%s] Down link stream %d does not exist, using 0",
939 info->streamName.c_str(), downStream );
940 downStream = 0;
941 }
942
943 //--------------------------------------------------------------------------
944 // Modify the message
945 //--------------------------------------------------------------------------
946 UnMarshallRequest( msg );
948 switch( hdr->requestid )
949 {
950 //------------------------------------------------------------------------
951 // Read - we update the path id to tell the server where we want to
952 // get the response, but we still send the request through stream 0
953 // We need to allocate space for read_args if we don't have it
954 // included yet
955 //------------------------------------------------------------------------
956 case kXR_read:
957 {
958 if( msg->GetSize() < sizeof(ClientReadRequest) + 8 )
959 {
960 msg->ReAllocate( sizeof(ClientReadRequest) + 8 );
961 void *newBuf = msg->GetBuffer(sizeof(ClientReadRequest));
962 memset( newBuf, 0, 8 );
964 req->dlen += 8;
965 }
966 read_args *args = (read_args*)msg->GetBuffer(sizeof(ClientReadRequest));
967 args->pathid = info->stream[downStream].pathId;
968 break;
969 }
970
971
972 //------------------------------------------------------------------------
973 // PgRead - we update the path id to tell the server where we want to
974 // get the response, but we still send the request through stream 0
975 // We need to allocate space for ClientPgReadReqArgs if we don't have it
976 // included yet
977 //------------------------------------------------------------------------
978 case kXR_pgread:
979 {
980 if( msg->GetSize() < sizeof( ClientPgReadRequest ) + sizeof( ClientPgReadReqArgs ) )
981 {
982 msg->ReAllocate( sizeof( ClientPgReadRequest ) + sizeof( ClientPgReadReqArgs ) );
983 void *newBuf = msg->GetBuffer( sizeof( ClientPgReadRequest ) );
984 memset( newBuf, 0, sizeof( ClientPgReadReqArgs ) );
986 req->dlen += sizeof( ClientPgReadReqArgs );
987 }
988 ClientPgReadReqArgs *args = reinterpret_cast<ClientPgReadReqArgs*>(
989 msg->GetBuffer( sizeof( ClientPgReadRequest ) ) );
990 args->pathid = info->stream[downStream].pathId;
991 break;
992 }
993
994 //------------------------------------------------------------------------
995 // ReadV - the situation is identical to read but we don't need any
996 // additional structures to specify the return path
997 //------------------------------------------------------------------------
998 case kXR_readv:
999 {
1001 req->pathid = info->stream[downStream].pathId;
1002 break;
1003 }
1004
1005 //------------------------------------------------------------------------
1006 // Write - multiplexing writes doesn't work properly in the server
1007 //------------------------------------------------------------------------
1008 case kXR_write:
1009 {
1010// ClientWriteRequest *req = (ClientWriteRequest*)msg->GetBuffer();
1011// req->pathid = info->stream[downStream].pathId;
1012 break;
1013 }
1014
1015 //------------------------------------------------------------------------
1016 // WriteV - multiplexing writes doesn't work properly in the server
1017 //------------------------------------------------------------------------
1018 case kXR_writev:
1019 {
1020// ClientWriteVRequest *req = (ClientWriteVRequest*)msg->GetBuffer();
1021// req->pathid = info->stream[downStream].pathId;
1022 break;
1023 }
1024
1025 //------------------------------------------------------------------------
1026 // PgWrite - multiplexing writes doesn't work properly in the server
1027 //------------------------------------------------------------------------
1028 case kXR_pgwrite:
1029 {
1030// ClientWriteVRequest *req = (ClientWriteVRequest*)msg->GetBuffer();
1031// req->pathid = info->stream[downStream].pathId;
1032 break;
1033 }
1034 };
1035 MarshallRequest( msg );
1036 return PathID( upStream, downStream );
1037 }
1038
1039 //----------------------------------------------------------------------------
1040 // Return a number of substreams per stream that should be created
1041 // This depends on the environment and whether we are connected to
1042 // a data server or not
1043 //----------------------------------------------------------------------------
1045 {
1046 XRootDChannelInfo *info = 0;
1047 channelData.Get( info );
1048
1049 if (!info) {
1050 DefaultEnv::GetLog()->Error(XRootDTransportMsg, "Internal error: no channel info");
1051 return 1;
1052 }
1053
1054 XrdSysMutexHelper scopedLock( info->mutex );
1055
1056 //--------------------------------------------------------------------------
1057 // If the connection has been opened in order to orchestrate a TPC or
1058 // the remote server is a Manager or Metamanager we will need only one
1059 // (control) stream.
1060 //--------------------------------------------------------------------------
1061 if( info->istpc || !(info->serverFlags & kXR_isServer ) ) return 1;
1062
1063 //--------------------------------------------------------------------------
1064 // Number of streams requested by user
1065 //--------------------------------------------------------------------------
1066 uint16_t ret = info->stream.size();
1067
1069 int nodata = DefaultTlsNoData;
1070 env->GetInt( "TlsNoData", nodata );
1071
1072 // Does the server require the stream 0 to be encrypted?
1073 bool srvTlsStrm0 = ( info->serverFlags & kXR_gotoTLS ) ||
1074 ( info->serverFlags & kXR_tlsLogin ) ||
1075 ( info->serverFlags & kXR_tlsSess );
1076 // Does the server NOT require the data streams to be encrypted?
1077 bool srvNoTlsData = !( info->serverFlags & kXR_tlsData );
1078 // Does the user require the stream 0 to be encrypted?
1079 bool usrTlsStrm0 = info->encrypted;
1080 // Does the user NOT require the data streams to be encrypted?
1081 bool usrNoTlsData = !info->encrypted || ( info->encrypted && nodata );
1082
1083 if( ( usrTlsStrm0 && usrNoTlsData && srvNoTlsData ) ||
1084 ( srvTlsStrm0 && srvNoTlsData && usrNoTlsData ) )
1085 {
1086 //------------------------------------------------------------------------
1087 // The server or user asked us to encrypt stream 0, but to send the data
1088 // (read/write) using a plain TCP connection
1089 //------------------------------------------------------------------------
1090 if( ret == 1 ) ++ret;
1091 }
1092
1093 if( ret > info->stream.size() )
1094 {
1095 info->stream.resize( ret );
1096 info->strmSelector->AdjustQueues( ret );
1097 }
1098
1099 return ret;
1100 }
1101
1102 //----------------------------------------------------------------------------
1103 // Marshall
1104 //----------------------------------------------------------------------------
1106 {
1107 ClientRequest *req = (ClientRequest*)msg;
1108 switch( req->header.requestid )
1109 {
1110 //------------------------------------------------------------------------
1111 // kXR_protocol
1112 //------------------------------------------------------------------------
1113 case kXR_protocol:
1114 req->protocol.clientpv = htonl( req->protocol.clientpv );
1115 break;
1116
1117 //------------------------------------------------------------------------
1118 // kXR_login
1119 //------------------------------------------------------------------------
1120 case kXR_login:
1121 req->login.pid = htonl( req->login.pid );
1122 break;
1123
1124 //------------------------------------------------------------------------
1125 // kXR_locate
1126 //------------------------------------------------------------------------
1127 case kXR_locate:
1128 req->locate.options = htons( req->locate.options );
1129 break;
1130
1131 //------------------------------------------------------------------------
1132 // kXR_query
1133 //------------------------------------------------------------------------
1134 case kXR_query:
1135 req->query.infotype = htons( req->query.infotype );
1136 break;
1137
1138 //------------------------------------------------------------------------
1139 // kXR_truncate
1140 //------------------------------------------------------------------------
1141 case kXR_truncate:
1142 req->truncate.offset = htonll( req->truncate.offset );
1143 break;
1144
1145 //------------------------------------------------------------------------
1146 // kXR_mkdir
1147 //------------------------------------------------------------------------
1148 case kXR_mkdir:
1149 req->mkdir.mode = htons( req->mkdir.mode );
1150 break;
1151
1152 //------------------------------------------------------------------------
1153 // kXR_chmod
1154 //------------------------------------------------------------------------
1155 case kXR_chmod:
1156 req->chmod.mode = htons( req->chmod.mode );
1157 break;
1158
1159 //------------------------------------------------------------------------
1160 // kXR_open
1161 //------------------------------------------------------------------------
1162 case kXR_open:
1163 req->open.mode = htons( req->open.mode );
1164 req->open.options = htons( req->open.options );
1165 req->open.optiont = htons( req->open.optiont );
1166 break;
1167
1168 //------------------------------------------------------------------------
1169 // kXR_read
1170 //------------------------------------------------------------------------
1171 case kXR_read:
1172 req->read.offset = htonll( req->read.offset );
1173 req->read.rlen = htonl( req->read.rlen );
1174 break;
1175
1176 //------------------------------------------------------------------------
1177 // kXR_write
1178 //------------------------------------------------------------------------
1179 case kXR_write:
1180 req->write.offset = htonll( req->write.offset );
1181 break;
1182
1183 //------------------------------------------------------------------------
1184 // kXR_mv
1185 //------------------------------------------------------------------------
1186 case kXR_mv:
1187 req->mv.arg1len = htons( req->mv.arg1len );
1188 break;
1189
1190 //------------------------------------------------------------------------
1191 // kXR_readv
1192 //------------------------------------------------------------------------
1193 case kXR_readv:
1194 {
1195 uint16_t numChunks = (req->readv.dlen)/16;
1196 readahead_list *dataChunk = (readahead_list*)( msg + 24 );
1197 for( size_t i = 0; i < numChunks; ++i )
1198 {
1199 dataChunk[i].rlen = htonl( dataChunk[i].rlen );
1200 dataChunk[i].offset = htonll( dataChunk[i].offset );
1201 }
1202 break;
1203 }
1204
1205 case kXR_clone:
1206 {
1207 uint32_t numChunks = (req->clone.dlen)/sizeof(XrdProto::clone_list);
1208 XrdProto::clone_list *dataChunk =
1209 (XrdProto::clone_list*)( msg + sizeof( ClientRequestHdr ) );
1210 for( size_t i = 0; i < numChunks; ++i )
1211 {
1212 dataChunk[i].srcOffs = htonll( dataChunk[i].srcOffs );
1213 dataChunk[i].srcLen = htonll( dataChunk[i].srcLen );
1214 dataChunk[i].dstOffs = htonll( dataChunk[i].dstOffs );
1215 }
1216 break;
1217 }
1218
1219 //------------------------------------------------------------------------
1220 // kXR_writev
1221 //------------------------------------------------------------------------
1222 case kXR_writev:
1223 {
1224 uint16_t numChunks = (req->writev.dlen)/16;
1225 XrdProto::write_list *wrtList =
1226 reinterpret_cast<XrdProto::write_list*>( msg + 24 );
1227 for( size_t i = 0; i < numChunks; ++i )
1228 {
1229 wrtList[i].wlen = htonl( wrtList[i].wlen );
1230 wrtList[i].offset = htonll( wrtList[i].offset );
1231 }
1232
1233 break;
1234 }
1235
1236 case kXR_pgread:
1237 {
1238 req->pgread.offset = htonll( req->pgread.offset );
1239 req->pgread.rlen = htonl( req->pgread.rlen );
1240 break;
1241 }
1242
1243 case kXR_pgwrite:
1244 {
1245 req->pgwrite.offset = htonll( req->pgwrite.offset );
1246 break;
1247 }
1248
1249 //------------------------------------------------------------------------
1250 // kXR_prepare
1251 //------------------------------------------------------------------------
1252 case kXR_prepare:
1253 {
1254 req->prepare.optionX = htons( req->prepare.optionX );
1255 req->prepare.port = htons( req->prepare.port );
1256 break;
1257 }
1258
1259 case kXR_chkpoint:
1260 {
1261 if( req->chkpoint.opcode == kXR_ckpXeq )
1262 MarshallRequest( msg + 24 );
1263 break;
1264 }
1265 };
1266
1267 req->header.requestid = htons( req->header.requestid );
1268 req->header.dlen = htonl( req->header.dlen );
1269 return XRootDStatus();
1270 }
1271
1272 //----------------------------------------------------------------------------
1273 // Unmarshall the request - sometimes the requests need to be rewritten,
1274 // so we need to unmarshall them
1275 //----------------------------------------------------------------------------
1277 {
1278 if( !msg->IsMarshalled() ) return XRootDStatus( stOK, suAlreadyDone );
1279 // We rely on the marshaling process to be symmetric!
1280 // First we unmarshall the request ID and the length because
1281 // MarshallRequest() relies on these, and then we need to unmarshall these
1282 // two again, because they get marshalled in MarshallRequest().
1283 // All this is pretty damn ugly and should be rewritten.
1284 ClientRequest *req = (ClientRequest*)msg->GetBuffer();
1285 req->header.requestid = htons( req->header.requestid );
1286 req->header.dlen = htonl( req->header.dlen );
1287 XRootDStatus st = MarshallRequest( msg );
1288 req->header.requestid = htons( req->header.requestid );
1289 req->header.dlen = htonl( req->header.dlen );
1290 msg->SetIsMarshalled( false );
1291 return st;
1292 }
1293
1294 //----------------------------------------------------------------------------
1295 // Unmarshall the body of the incoming message
1296 //----------------------------------------------------------------------------
1298 {
1300
1301 //--------------------------------------------------------------------------
1302 // kXR_ok
1303 //--------------------------------------------------------------------------
1304 if( m->hdr.status == kXR_ok )
1305 {
1306 switch( reqType )
1307 {
1308 //----------------------------------------------------------------------
1309 // kXR_protocol
1310 //----------------------------------------------------------------------
1311 case kXR_protocol:
1312 if( m->hdr.dlen < 8 )
1313 return XRootDStatus( stError, errInvalidMessage, 0, "kXR_protocol: body too short." );
1314 m->body.protocol.pval = ntohl( m->body.protocol.pval );
1315 m->body.protocol.flags = ntohl( m->body.protocol.flags );
1316 break;
1317 }
1318 }
1319 //--------------------------------------------------------------------------
1320 // kXR_error
1321 //--------------------------------------------------------------------------
1322 else if( m->hdr.status == kXR_error )
1323 {
1324 if( m->hdr.dlen < 4 )
1325 return XRootDStatus( stError, errInvalidMessage, 0, "kXR_error: body too short." );
1326 m->body.error.errnum = ntohl( m->body.error.errnum );
1327 }
1328
1329 //--------------------------------------------------------------------------
1330 // kXR_wait
1331 //--------------------------------------------------------------------------
1332 else if( m->hdr.status == kXR_wait )
1333 {
1334 if( m->hdr.dlen < 4 )
1335 return XRootDStatus( stError, errInvalidMessage, 0, "kXR_wait: body too short." );
1336 m->body.wait.seconds = htonl( m->body.wait.seconds );
1337 }
1338
1339 //--------------------------------------------------------------------------
1340 // kXR_redirect
1341 //--------------------------------------------------------------------------
1342 else if( m->hdr.status == kXR_redirect )
1343 {
1344 if( m->hdr.dlen < 4 )
1345 return XRootDStatus( stError, errInvalidMessage, 0, "kXR_redirect: body too short." );
1346 m->body.redirect.port = htonl( m->body.redirect.port );
1347 }
1348
1349 //--------------------------------------------------------------------------
1350 // kXR_waitresp
1351 //--------------------------------------------------------------------------
1352 else if( m->hdr.status == kXR_waitresp )
1353 {
1354 if( m->hdr.dlen < 4 )
1355 return XRootDStatus( stError, errInvalidMessage, 0, "kXR_waitresp: body too short." );
1356 m->body.waitresp.seconds = htonl( m->body.waitresp.seconds );
1357 }
1358
1359 //--------------------------------------------------------------------------
1360 // kXR_attn
1361 //--------------------------------------------------------------------------
1362 else if( m->hdr.status == kXR_attn )
1363 {
1364 if( m->hdr.dlen < 4 )
1365 return XRootDStatus( stError, errInvalidMessage, 0, "kXR_attn: body too short." );
1366 m->body.attn.actnum = htonl( m->body.attn.actnum );
1367 }
1368
1369 return XRootDStatus();
1370 }
1371
1372 //------------------------------------------------------------------------
1374 //------------------------------------------------------------------------
1376 {
1377 //--------------------------------------------------------------------------
1378 // Calculate the crc32c before the unmarshaling the body!
1379 //--------------------------------------------------------------------------
1381 char *buffer = msg.GetBuffer( 8 + sizeof( rspst->bdy.crc32c ) );
1382 size_t length = rspst->hdr.dlen - sizeof( rspst->bdy.crc32c );
1383 uint32_t crcval = XrdOucCRC::Calc32C( buffer, length );
1384
1385 size_t stlen = sizeof( ServerResponseStatus );
1386 switch( reqType )
1387 {
1388 case kXR_pgread:
1389 {
1390 stlen += sizeof( ServerResponseBody_pgRead );
1391 break;
1392 }
1393
1394 case kXR_pgwrite:
1395 {
1396 stlen += sizeof( ServerResponseBody_pgWrite );
1397 break;
1398 }
1399 }
1400
1401 if( msg.GetSize() < stlen ) return XRootDStatus( stError, errInvalidMessage, 0,
1402 "kXR_status: invalid message size." );
1403
1404 rspst->bdy.crc32c = ntohl( rspst->bdy.crc32c );
1405 rspst->bdy.dlen = ntohl( rspst->bdy.dlen );
1406
1407 switch( reqType )
1408 {
1409 case kXR_pgread:
1410 {
1412 pgrdbdy->offset = ntohll( pgrdbdy->offset );
1413 break;
1414 }
1415
1416 case kXR_pgwrite:
1417 {
1419 pgwrtbdy->offset = ntohll( pgwrtbdy->offset );
1420 break;
1421 }
1422 }
1423
1424 //--------------------------------------------------------------------------
1425 // Do the integrity checks
1426 //--------------------------------------------------------------------------
1427 if( crcval != rspst->bdy.crc32c )
1428 {
1429 return XRootDStatus( stError, errDataError, 0, "kXR_status response header "
1430 "corrupted (crc32c integrity check failed)." );
1431 }
1432
1433 if( rspst->hdr.streamid[0] != rspst->bdy.streamID[0] ||
1434 rspst->hdr.streamid[1] != rspst->bdy.streamID[1] )
1435 {
1436 return XRootDStatus( stError, errDataError, 0, "response header corrupted "
1437 "(stream ID mismatch)." );
1438 }
1439
1440
1441
1442 if( rspst->bdy.requestid + kXR_1stRequest != reqType )
1443 {
1444 return XRootDStatus( stError, errDataError, 0, "kXR_status response header corrupted "
1445 "(request ID mismatch)." );
1446 }
1447
1448 return XRootDStatus();
1449 }
1450
1452 {
1454 uint16_t reqType = rsp->status.bdy.requestid + kXR_1stRequest;
1455
1456 switch( reqType )
1457 {
1458 case kXR_pgwrite:
1459 {
1460 //--------------------------------------------------------------------------
1461 // If there's no additional data there's nothing to unmarshal
1462 //--------------------------------------------------------------------------
1463 if( rsp->status.bdy.dlen == 0 ) return XRootDStatus();
1464 //--------------------------------------------------------------------------
1465 // If there's not enough data to form correction-segment report an error
1466 //--------------------------------------------------------------------------
1467 if( size_t( rsp->status.bdy.dlen ) < sizeof( ServerResponseBody_pgWrCSE ) )
1469 "kXR_status: invalid message size." );
1470
1471 //--------------------------------------------------------------------------
1472 // Calculate the crc32c for the additional data
1473 //--------------------------------------------------------------------------
1475 cse->cseCRC = ntohl( cse->cseCRC );
1476 size_t length = rsp->status.bdy.dlen - sizeof( uint32_t );
1477 void* buffer = msg.GetBuffer( sizeof( ServerResponseV2 ) + sizeof( uint32_t ) );
1478 uint32_t crcval = XrdOucCRC::Calc32C( buffer, length );
1479
1480 //--------------------------------------------------------------------------
1481 // Do the integrity checks
1482 //--------------------------------------------------------------------------
1483 if( crcval != cse->cseCRC )
1484 {
1485 return XRootDStatus( stError, errDataError, 0, "kXR_status response header "
1486 "corrupted (crc32c integrity check failed)." );
1487 }
1488
1489 cse->dlFirst = ntohs( cse->dlFirst );
1490 cse->dlLast = ntohs( cse->dlLast );
1491
1492 size_t pgcnt = ( rsp->status.bdy.dlen - sizeof( ServerResponseBody_pgWrCSE ) ) /
1493 sizeof( kXR_int64 );
1494 kXR_int64 *pgoffs = (kXR_int64*)msg.GetBuffer( sizeof( ServerResponseV2 ) +
1495 sizeof( ServerResponseBody_pgWrCSE ) );
1496
1497 for( size_t i = 0; i < pgcnt; ++i )
1498 pgoffs[i] = ntohll( pgoffs[i] );
1499
1500 return XRootDStatus();
1501 break;
1502 }
1503
1504 default:
1505 break;
1506 }
1507
1509 }
1510
1511 //----------------------------------------------------------------------------
1512 // Unmarshall the header of the incoming message
1513 //----------------------------------------------------------------------------
1515 {
1517 header->status = ntohs( header->status );
1518 header->dlen = ntohl( header->dlen );
1519 }
1520
1521 //----------------------------------------------------------------------------
1522 // Log server error response
1523 //----------------------------------------------------------------------------
1525 {
1526 Log *log = DefaultEnv::GetLog();
1527 ServerResponse *rsp = (ServerResponse *)msg.GetBuffer();
1528 char *errmsg = new char[rsp->hdr.dlen-3]; errmsg[rsp->hdr.dlen-4] = 0;
1529 memcpy( errmsg, rsp->body.error.errmsg, rsp->hdr.dlen-4 );
1530 log->Error( XRootDTransportMsg, "Server responded with an error [%d]: %s",
1531 rsp->body.error.errnum, errmsg );
1532 delete [] errmsg;
1533 }
1534
1535 //------------------------------------------------------------------------
1536 // Number of currently connected data streams
1537 //------------------------------------------------------------------------
1539 {
1540 XRootDChannelInfo *info = 0;
1541 channelData.Get( info );
1542
1543 if (!info) {
1544 DefaultEnv::GetLog()->Error(XRootDTransportMsg, "Internal error: no channel info");
1545 return 0;
1546 }
1547
1548 XrdSysMutexHelper scopedLock( info->mutex );
1549
1550 uint16_t nbConnected = 0;
1551 for( size_t i = 1; i < info->stream.size(); ++i )
1552 if( info->stream[i].status == XRootDStreamInfo::Connected )
1553 ++nbConnected;
1554
1555 return nbConnected;
1556 }
1557
1558 //----------------------------------------------------------------------------
1559 // The stream has been disconnected, do the cleanups
1560 //----------------------------------------------------------------------------
1562 uint16_t subStreamId )
1563 {
1564 XRootDChannelInfo *info = 0;
1565 channelData.Get( info );
1566
1567 if (!info) {
1568 DefaultEnv::GetLog()->Error(XRootDTransportMsg, "Internal error: no channel info");
1569 return;
1570 }
1571
1572 XrdSysMutexHelper scopedLock( info->mutex );
1573
1574 if( !info->stream.empty() )
1575 {
1576 XRootDStreamInfo &sInfo = info->stream[subStreamId];
1578 }
1579
1580 if( subStreamId == 0 )
1581 {
1582 CleanUpProtection( info );
1583 info->sidManager->ReleaseAllTimedOut();
1584 info->sentOpens.clear();
1585 info->sentCloses.clear();
1586 info->openFiles = 0;
1587 info->waitBarrier = 0;
1588 }
1589 }
1590
1591 //------------------------------------------------------------------------
1592 // Query the channel
1593 //------------------------------------------------------------------------
1595 AnyObject &result,
1596 AnyObject &channelData )
1597 {
1598 XRootDChannelInfo *info = 0;
1599 channelData.Get( info );
1600
1601 if (!info)
1603
1604 XrdSysMutexHelper scopedLock( info->mutex );
1605
1606 switch( query )
1607 {
1608 //------------------------------------------------------------------------
1609 // Protocol name
1610 //------------------------------------------------------------------------
1612 result.Set( (const char*)"XRootD", false );
1613 return Status();
1614
1615 //------------------------------------------------------------------------
1616 // Authentication
1617 //------------------------------------------------------------------------
1619 result.Set( new std::string( info->authProtocolName ), false );
1620 return Status();
1621
1622 //------------------------------------------------------------------------
1623 // Server flags
1624 //------------------------------------------------------------------------
1626 result.Set( new int( info->serverFlags ), false );
1627 return Status();
1628
1629 //------------------------------------------------------------------------
1630 // Protocol version
1631 //------------------------------------------------------------------------
1633 result.Set( new int( info->protocolVersion ), false );
1634 return Status();
1635
1637 result.Set( new bool( info->encrypted ), false );
1638 return Status();
1639 };
1641 }
1642
1643 //----------------------------------------------------------------------------
1644 // Check whether the transport can hijack the message
1645 //----------------------------------------------------------------------------
1647 uint16_t subStream,
1648 AnyObject &channelData )
1649 {
1650 XRootDChannelInfo *info = 0;
1651 channelData.Get( info );
1652 if( !info ) return NoAction;
1653 XrdSysMutexHelper scopedLock( info->mutex );
1654 Log *log = DefaultEnv::GetLog();
1655
1656 //--------------------------------------------------------------------------
1657 // Update the substream queues
1658 //--------------------------------------------------------------------------
1659 info->strmSelector->MsgReceived( subStream );
1660
1661 //--------------------------------------------------------------------------
1662 // Check whether this message is a response to a request that has
1663 // timed out, and if so, drop it
1664 //--------------------------------------------------------------------------
1666 if( rsp->hdr.status == kXR_attn )
1667 {
1668 return NoAction;
1669 }
1670
1671 if( info->sidManager->IsTimedOut( rsp->hdr.streamid ) )
1672 {
1673 log->Error( XRootDTransportMsg, "Message %p, stream [%d, %d] is a "
1674 "response that we're no longer interested in (timed out)",
1675 (void*)&msg, rsp->hdr.streamid[0], rsp->hdr.streamid[1] );
1676 //------------------------------------------------------------------------
1677 // If it is kXR_waitresp there will be another one,
1678 // so we don't release the sid yet
1679 //------------------------------------------------------------------------
1680 if( rsp->hdr.status != kXR_waitresp )
1681 info->sidManager->ReleaseTimedOut( rsp->hdr.streamid );
1682 //------------------------------------------------------------------------
1683 // If it is a successful response to an open request
1684 // that timed out, we need to send a close
1685 //------------------------------------------------------------------------
1686 uint16_t sid; memcpy( &sid, rsp->hdr.streamid, 2 );
1687 std::set<uint16_t>::iterator sidIt = info->sentOpens.find( sid );
1688 if( sidIt != info->sentOpens.end() )
1689 {
1690 info->sentOpens.erase( sidIt );
1691 if( rsp->hdr.status == kXR_ok ) return RequestClose;
1692 }
1693 return DigestMsg;
1694 }
1695
1696 //--------------------------------------------------------------------------
1697 // If we have a wait or waitresp
1698 //--------------------------------------------------------------------------
1699 uint32_t seconds = 0;
1700 if( rsp->hdr.status == kXR_wait )
1701 seconds = ntohl( rsp->body.wait.seconds ) + 5; // we need extra time
1702 // to re-send the request
1703 else if( rsp->hdr.status == kXR_waitresp )
1704 {
1705 seconds = ntohl( rsp->body.waitresp.seconds );
1706
1707 log->Dump( XRootDMsg, "[%s] Got kXR_waitresp response of %u seconds, "
1708 "setting up wait barrier.",
1709 info->streamName.c_str(),
1710 seconds );
1711 }
1712
1713 time_t barrier = time(0) + seconds;
1714 if( info->waitBarrier < barrier )
1715 info->waitBarrier = barrier;
1716
1717 //--------------------------------------------------------------------------
1718 // If we got a response to an open request, we may need to bump the counter
1719 // of open files
1720 //--------------------------------------------------------------------------
1721 uint16_t sid; memcpy( &sid, rsp->hdr.streamid, 2 );
1722 std::set<uint16_t>::iterator sidIt = info->sentOpens.find( sid );
1723 if( sidIt != info->sentOpens.end() )
1724 {
1725 if( rsp->hdr.status == kXR_waitresp )
1726 return NoAction;
1727 info->sentOpens.erase( sidIt );
1728 if( rsp->hdr.status == kXR_ok )
1729 {
1730 ++info->openFiles;
1731 info->finstcnt.fetch_add( 1, std::memory_order_relaxed ); // another file File object instance has been bound with this connection
1732 }
1733 return NoAction;
1734 }
1735
1736 //--------------------------------------------------------------------------
1737 // If we got a response to a close, we may need to decrement the counter of
1738 // open files
1739 //--------------------------------------------------------------------------
1740 sidIt = info->sentCloses.find( sid );
1741 if( sidIt != info->sentCloses.end() )
1742 {
1743 if( rsp->hdr.status == kXR_waitresp )
1744 return NoAction;
1745 info->sentCloses.erase( sidIt );
1746 --info->openFiles;
1747 return NoAction;
1748 }
1749 return NoAction;
1750 }
1751
1752 //----------------------------------------------------------------------------
1753 // Notify the transport about a message having been sent
1754 //----------------------------------------------------------------------------
1756 uint16_t subStream,
1757 uint32_t bytesSent,
1758 AnyObject &channelData )
1759 {
1760 // Called when a message has been sent. For messages that return on a
1761 // different pathid (and hence may use a different poller) it is possible
1762 // that the server has already replied and the reply will trigger
1763 // MessageReceived() before this method has been called. However for open
1764 // and close this is never the case and this method is used for tracking
1765 // only those.
1766 XRootDChannelInfo *info = 0;
1767 channelData.Get( info );
1768 if( !info ) return;
1769 XrdSysMutexHelper scopedLock( info->mutex );
1770 ClientRequest *req = (ClientRequest*)msg->GetBuffer();
1771 uint16_t reqid = ntohs( req->header.requestid );
1772
1773
1774 //--------------------------------------------------------------------------
1775 // We need to track opens to know if we can close streams due to idleness
1776 //--------------------------------------------------------------------------
1777 uint16_t sid;
1778 memcpy( &sid, req->header.streamid, 2 );
1779
1780 if( reqid == kXR_open )
1781 info->sentOpens.insert( sid );
1782 else if( reqid == kXR_close )
1783 info->sentCloses.insert( sid );
1784 }
1785
1786
1787 //----------------------------------------------------------------------------
1788 // Get signature for given message
1789 //----------------------------------------------------------------------------
1791 {
1792 XRootDChannelInfo *info = 0;
1793 channelData.Get( info );
1794 return GetSignature( toSign, sign, info );
1795 }
1796
1797 //------------------------------------------------------------------------
1799 //------------------------------------------------------------------------
1801 Message *&sign,
1802 XRootDChannelInfo *info )
1803 {
1804 XrdSysRWLockHelper scope( pSecUnloadHandler->lock );
1805 if( pSecUnloadHandler->unloaded ) return Status( stError, errInvalidOp );
1806
1807 ClientRequest *thereq = reinterpret_cast<ClientRequest*>( toSign->GetBuffer() );
1808 if( !info ) return Status( stError, errInternal );
1809 if( info->protection )
1810 {
1811 SecurityRequest *newreq = 0;
1812 // check if we have to secure the request in the first place
1813 if( !( NEED2SECURE ( info->protection )( *thereq ) ) ) return Status();
1814 // secure (sign/encrypt) the request
1815 int rc = info->protection->Secure( newreq, *thereq, 0 );
1816 // there was an error
1817 if( rc < 0 )
1818 return Status( stError, errInternal, -rc );
1819
1820 sign = new Message();
1821 sign->Grab( reinterpret_cast<char*>( newreq ), rc );
1822 }
1823
1824 return Status();
1825 }
1826
1827 //------------------------------------------------------------------------
1829 //------------------------------------------------------------------------
1831 {
1832 XRootDChannelInfo *info = 0;
1833 channelData.Get( info );
1834 if( info->finstcnt.load( std::memory_order_relaxed ) > 0 )
1835 info->finstcnt.fetch_sub( 1, std::memory_order_relaxed );
1836 }
1837
1838 //----------------------------------------------------------------------------
1839 // Wait before exit
1840 //----------------------------------------------------------------------------
1842 {
1843 XrdSysRWLockHelper scope( pSecUnloadHandler->lock, false ); // obtain write lock
1844 pSecUnloadHandler->unloaded = true;
1845 }
1846
1847 //----------------------------------------------------------------------------
1848 // @return : true if encryption should be turned on, false otherwise
1849 //----------------------------------------------------------------------------
1851 AnyObject &channelData )
1852 {
1853 XRootDChannelInfo *info = 0;
1854 channelData.Get( info );
1855
1857 int notlsok = DefaultNoTlsOK;
1858 env->GetInt( "NoTlsOK", notlsok );
1859
1860
1861 if( notlsok )
1862 return info->encrypted;
1863
1864 XRootDStreamInfo &sInfo = info->stream[handShakeData->subStreamId];
1865
1866 // Did the server instructed us to switch to TLS right away?
1867 if( sInfo.serverFlags & kXR_gotoTLS )
1868 {
1869 if( handShakeData->subStreamId == 0 ) info->encrypted = true;
1870 return true ;
1871 }
1872
1873 //--------------------------------------------------------------------------
1874 // The control stream (sub-stream 0) might need to switch to TLS before
1875 // login or after login
1876 //--------------------------------------------------------------------------
1877 if( handShakeData->subStreamId == 0 )
1878 {
1879 //------------------------------------------------------------------------
1880 // We are about to login and the server asked to start encrypting
1881 // before login
1882 //------------------------------------------------------------------------
1883 if( ( sInfo.status == XRootDStreamInfo::LoginSent ) &&
1884 ( info->serverFlags & kXR_tlsLogin ) )
1885 {
1886 info->encrypted = true;
1887 return true;
1888 }
1889
1890 //--------------------------------------------------------------------
1891 // The hand-shake is done and the server requested to encrypt the session
1892 //--------------------------------------------------------------------
1893 if( (sInfo.status == XRootDStreamInfo::Connected ||
1894 //--------------------------------------------------------------------
1895 // we really need to turn on TLS before we sent kXR_endsess and we
1896 // are about to do so (1st enable encryption, then send kXR_endsess)
1897 //--------------------------------------------------------------------
1899 ( info->serverFlags & kXR_tlsSess ) )
1900 {
1901 info->encrypted = true;
1902 return true;
1903 }
1904 }
1905 //--------------------------------------------------------------------------
1906 // A data stream (sub-stream > 0) if need be will be switched to TLS before
1907 // bind.
1908 //--------------------------------------------------------------------------
1909 else
1910 {
1911 //------------------------------------------------------------------------
1912 // We are about to bind a data stream and the server asked to start
1913 // encrypting before bind
1914 //------------------------------------------------------------------------
1915 if( ( sInfo.status == XRootDStreamInfo::BindSent ) &&
1916 ( info->serverFlags & kXR_tlsData ) )
1917 {
1918 return true;
1919 }
1920 }
1921
1922 return false;
1923 }
1924
1925 //------------------------------------------------------------------------
1926 // Get bind preference for the next data stream
1927 //------------------------------------------------------------------------
1929 AnyObject &channelData )
1930 {
1931 XRootDChannelInfo *info = 0;
1932 channelData.Get( info );
1933
1934 if(!info || !info->bindSelector)
1935 return url;
1936
1937 return URL( info->bindSelector->Get() );
1938 }
1939
1940 //----------------------------------------------------------------------------
1941 // Generate the message to be sent as an initial handshake
1942 // (handshake+kXR_protocol)
1943 //----------------------------------------------------------------------------
1944 Message *XRootDTransport::GenerateInitialHSProtocol( HandShakeData *hsData,
1945 XRootDChannelInfo *info,
1946 kXR_char expect )
1947 {
1948 Log *log = DefaultEnv::GetLog();
1950 "[%s] Sending out the initial hand shake + kXR_protocol",
1951 hsData->streamName.c_str() );
1952
1953 Message *msg = new Message();
1954
1955 msg->Allocate( 20+sizeof(ClientProtocolRequest) );
1956 msg->Zero();
1957
1959 init->fourth = htonl(4);
1960 init->fifth = htonl(2012);
1961
1963 InitProtocolReq( proto, info, expect );
1964
1965 return msg;
1966 }
1967
1968 //------------------------------------------------------------------------
1969 // Generate the protocol message
1970 //------------------------------------------------------------------------
1971 Message *XRootDTransport::GenerateProtocol( HandShakeData *hsData,
1972 XRootDChannelInfo *info,
1973 kXR_char expect )
1974 {
1975 Log *log = DefaultEnv::GetLog();
1976 log->Debug( XRootDTransportMsg,
1977 "[%s] Sending out the kXR_protocol",
1978 hsData->streamName.c_str() );
1979
1980 Message *msg = new Message();
1981 msg->Allocate( sizeof(ClientProtocolRequest) );
1982 msg->Zero();
1983
1984 ClientProtocolRequest *proto = (ClientProtocolRequest *)msg->GetBuffer();
1985 InitProtocolReq( proto, info, expect );
1986
1987 return msg;
1988 }
1989
1990 //------------------------------------------------------------------------
1991 // Initialize protocol request
1992 //------------------------------------------------------------------------
1993 void XRootDTransport::InitProtocolReq( ClientProtocolRequest *request,
1994 XRootDChannelInfo *info,
1995 kXR_char expect )
1996 {
1997 request->requestid = htons(kXR_protocol);
1998 request->clientpv = htonl(kXR_PROTOCOLVERSION);
2001
2002 int notlsok = DefaultNoTlsOK;
2003 int tlsnodata = DefaultTlsNoData;
2004
2005 XrdCl::Env *env = XrdCl::DefaultEnv::GetEnv();
2006
2007 env->GetInt( "NoTlsOK", notlsok );
2008
2010 env->GetInt( "TlsNoData", tlsnodata );
2011
2012 if (info->encrypted || InitTLS())
2014
2015 if (info->encrypted && !(notlsok || tlsnodata))
2017
2018 request->expect = expect;
2019
2020 //--------------------------------------------------------------------------
2021 // If we are in the curse of establishing a connection in the context of
2022 // TPC update the expect! (this will be never followed be a bind)
2023 //--------------------------------------------------------------------------
2024 if( info->istpc )
2026 }
2027
2028 //----------------------------------------------------------------------------
2029 // Process the server initial handshake response
2030 //----------------------------------------------------------------------------
2031 XRootDStatus XRootDTransport::ProcessServerHS( HandShakeData *hsData,
2032 XRootDChannelInfo *info )
2033 {
2034 Log *log = DefaultEnv::GetLog();
2035
2036 Message *msg = hsData->in;
2037 ServerResponseHeader *respHdr = (ServerResponseHeader *)msg->GetBuffer();
2038 ServerInitHandShake *hs = (ServerInitHandShake *)msg->GetBuffer(4);
2039
2040 if( respHdr->status != kXR_ok )
2041 {
2042 log->Error( XRootDTransportMsg, "[%s] Invalid hand shake response",
2043 hsData->streamName.c_str() );
2044
2045 return XRootDStatus( stFatal, errHandShakeFailed, 0, "Invalid hand shake response." );
2046 }
2047
2048 XRootDStreamInfo &sInfo = info->stream[hsData->subStreamId];
2049 const uint32_t pv = ntohl(hs->protover);
2050 sInfo.serverFlags = ntohl(hs->msgval) == kXR_DataServer ?
2053
2054 if( hsData->subStreamId == 0 )
2055 {
2056 info->protocolVersion = pv;
2057 info->serverFlags = sInfo.serverFlags;
2058 }
2059
2060 log->Debug( XRootDTransportMsg,
2061 "[%s] Got the server hand shake response (%s, protocol "
2062 "version %x)",
2063 hsData->streamName.c_str(),
2064 ServerFlagsToStr( sInfo.serverFlags ).c_str(),
2065 info->protocolVersion );
2066
2067 return XRootDStatus( stOK, suContinue );
2068 }
2069
2070 //----------------------------------------------------------------------------
2071 // Process the protocol response
2072 //----------------------------------------------------------------------------
2073 XRootDStatus XRootDTransport::ProcessProtocolResp( HandShakeData *hsData,
2074 XRootDChannelInfo *info )
2075 {
2076 Log *log = DefaultEnv::GetLog();
2077
2078 XRootDStatus st = UnMarshallBody( hsData->in, kXR_protocol );
2079 if( !st.IsOK() )
2080 return st;
2081
2082 ServerResponse *rsp = (ServerResponse*)hsData->in->GetBuffer();
2083
2084
2085 if( rsp->hdr.status != kXR_ok )
2086 {
2087 log->Error( XRootDTransportMsg, "[%s] kXR_protocol request failed",
2088 hsData->streamName.c_str() );
2089
2090 return XRootDStatus( stFatal, errHandShakeFailed, 0, "kXR_protocol request failed" );
2091 }
2092
2093 XRootDStreamInfo &sInfo = info->stream[hsData->subStreamId];
2094 if( rsp->body.protocol.pval >= 0x297 )
2095 sInfo.serverFlags = rsp->body.protocol.flags;
2096
2097 if( hsData->subStreamId > 0 )
2098 return XRootDStatus( stOK, suContinue );
2099
2100 info->serverFlags = sInfo.serverFlags;
2101
2102 XrdCl::Env *env = XrdCl::DefaultEnv::GetEnv();
2103 int notlsok = DefaultNoTlsOK;
2104 env->GetInt( "NoTlsOK", notlsok );
2105
2106 if( rsp->body.protocol.pval < kXR_PROTTLSVERSION && info->encrypted )
2107 {
2108 //------------------------------------------------------------------------
2109 // User requested an encrypted connection but the server is to old to
2110 // support it!
2111 //------------------------------------------------------------------------
2112 if( !notlsok ) return XRootDStatus( stFatal, errTlsError, ENOTSUP, "TLS not supported" );
2113
2114 //------------------------------------------------------------------------
2115 // We are falling back to unencrypted data transmission, as configured
2116 // in XRD_NOTLSOK environment variable
2117 //------------------------------------------------------------------------
2118 log->Info( XRootDTransportMsg,
2119 "[%s] Falling back to unencrypted transmission, server does "
2120 "not support TLS encryption.",
2121 hsData->streamName.c_str() );
2122 info->encrypted = false;
2123 }
2124
2125 if( rsp->body.protocol.pval >= 0x297 )
2126 info->serverFlags = rsp->body.protocol.flags;
2127
2128 if( rsp->hdr.dlen > 8 )
2129 {
2130 info->protRespBody = new ServerResponseBody_Protocol();
2131 info->protRespBody->flags = rsp->body.protocol.flags;
2132 info->protRespBody->pval = rsp->body.protocol.pval;
2133
2134 char* bodybuff = reinterpret_cast<char*>( &rsp->body.protocol.secreq );
2135 size_t bodysize = rsp->hdr.dlen - 8;
2136 XRootDStatus st = ProcessProtocolBody( bodybuff, bodysize, info );
2137 if( !st.IsOK() )
2138 return st;
2139 }
2140
2141 log->Debug( XRootDTransportMsg,
2142 "[%s] kXR_protocol successful (%s, protocol version %x)",
2143 hsData->streamName.c_str(),
2144 ServerFlagsToStr( info->serverFlags ).c_str(),
2145 info->protocolVersion );
2146
2147 if( !( info->serverFlags & kXR_haveTLS ) && info->encrypted )
2148 {
2149 //------------------------------------------------------------------------
2150 // User requested an encrypted connection but the server was not configured
2151 // to support encryption!
2152 //------------------------------------------------------------------------
2153 return XRootDStatus( stFatal, errTlsError, ECONNREFUSED,
2154 "Server was not configured to support encryption." );
2155 }
2156
2157 //--------------------------------------------------------------------------
2158 // Now see if we have to enforce encryption in case the server does not
2159 // support PgRead/PgWrite
2160 //--------------------------------------------------------------------------
2161 int tlsOnNoPgrw = DefaultWantTlsOnNoPgrw;
2162 env->GetInt( "WantTlsOnNoPgrw", tlsOnNoPgrw );
2163 if( !( info->serverFlags & kXR_suppgrw ) && tlsOnNoPgrw )
2164 {
2165 //------------------------------------------------------------------------
2166 // If user requested encryption just make sure it is not switched off for
2167 // data
2168 //------------------------------------------------------------------------
2169 if( info->encrypted )
2170 {
2171 log->Debug( XRootDTransportMsg,
2172 "[%s] Server does not support PgRead/PgWrite and"
2173 " WantTlsOnNoPgrw is on; enforcing encryption for data.",
2174 hsData->streamName.c_str() );
2175 env->PutInt( "TlsNoData", DefaultTlsNoData );
2176 }
2177 //------------------------------------------------------------------------
2178 // Otherwise, if server is not enforcing data encryption, we will need to
2179 // redo the protocol request with kXR_wantTLS set.
2180 //------------------------------------------------------------------------
2181 else if( !( info->serverFlags & kXR_tlsData ) &&
2182 ( info->serverFlags & kXR_haveTLS ) )
2183 {
2184 info->encrypted = true;
2185 return XRootDStatus( stOK, suRetry );
2186 }
2187 }
2188
2189 return XRootDStatus( stOK, suContinue );
2190 }
2191
2192 XRootDStatus XRootDTransport::ProcessProtocolBody( char *bodybuff,
2193 size_t bodysize,
2194 XRootDChannelInfo *info )
2195 {
2196 //--------------------------------------------------------------------------
2197 // Parse bind preferences
2198 //--------------------------------------------------------------------------
2199 XrdProto::bifReqs *bifreq = reinterpret_cast<XrdProto::bifReqs*>( bodybuff );
2200 if( bodysize >= sizeof( XrdProto::bifReqs ) && bifreq->theTag == 'B' )
2201 {
2202 bodybuff += sizeof( XrdProto::bifReqs );
2203 bodysize -= sizeof( XrdProto::bifReqs );
2204
2205 if( bodysize < bifreq->bifILen )
2206 return XRootDStatus( stError, errDataError, 0, "Received incomplete "
2207 "protocol response." );
2208 std::string bindprefs_str( bodybuff, bifreq->bifILen );
2209 std::vector<std::string> bindprefs;
2210 Utils::splitString( bindprefs, bindprefs_str, "," );
2211 info->bindSelector.reset( new BindPrefSelector( std::move( bindprefs ) ) );
2212 bodybuff += bifreq->bifILen;
2213 bodysize -= bifreq->bifILen;
2214 }
2215 //--------------------------------------------------------------------------
2216 // Parse security requirements
2217 //--------------------------------------------------------------------------
2218 XrdProto::secReqs *secreq = reinterpret_cast<XrdProto::secReqs*>( bodybuff );
2219 if( bodysize >= 6 /*XrdProto::secReqs*/ && secreq->theTag == 'S' )
2220 {
2221 memcpy( &info->protRespBody->secreq, secreq, bodysize );
2222 info->protRespSize = bodysize + 8 /*pval & flags*/;
2223 }
2224
2225 return XRootDStatus();
2226 }
2227
2228 //----------------------------------------------------------------------------
2229 // Generate the bind message
2230 //----------------------------------------------------------------------------
2231 Message *XRootDTransport::GenerateBind( HandShakeData *hsData,
2232 XRootDChannelInfo *info )
2233 {
2234 Log *log = DefaultEnv::GetLog();
2235
2236 log->Debug( XRootDTransportMsg,
2237 "[%s] Sending out the bind request",
2238 hsData->streamName.c_str() );
2239
2240
2241 Message *msg = new Message( sizeof( ClientBindRequest ) );
2242 ClientBindRequest *bindReq = (ClientBindRequest *)msg->GetBuffer();
2243
2244 bindReq->requestid = kXR_bind;
2245 memcpy( bindReq->sessid, info->sessionId, 16 );
2246 bindReq->dlen = 0;
2247 MarshallRequest( msg );
2248 return msg;
2249 }
2250
2251 //----------------------------------------------------------------------------
2252 // Generate the bind message
2253 //----------------------------------------------------------------------------
2254 XRootDStatus XRootDTransport::ProcessBindResp( HandShakeData *hsData,
2255 XRootDChannelInfo *info )
2256 {
2257 Log *log = DefaultEnv::GetLog();
2258
2259 XRootDStatus st = UnMarshallBody( hsData->in, kXR_bind );
2260 if( !st.IsOK() )
2261 return st;
2262
2263 ServerResponse *rsp = (ServerResponse*)hsData->in->GetBuffer();
2264
2265 if( rsp->hdr.status != kXR_ok )
2266 {
2267 log->Error( XRootDTransportMsg, "[%s] kXR_bind request failed",
2268 hsData->streamName.c_str() );
2269 return XRootDStatus( stFatal, errHandShakeFailed, 0, "kXR_bind request failed" );
2270 }
2271
2272 info->stream[hsData->subStreamId].pathId = rsp->body.bind.substreamid;
2273 log->Debug( XRootDTransportMsg, "[%s] kXR_bind successful",
2274 hsData->streamName.c_str() );
2275
2276 return XRootDStatus();
2277 }
2278
2279 //----------------------------------------------------------------------------
2280 // Generate the login message
2281 //----------------------------------------------------------------------------
2282 Message *XRootDTransport::GenerateLogIn( HandShakeData *hsData,
2283 XRootDChannelInfo *info )
2284 {
2285 Log *log = DefaultEnv::GetLog();
2286 Env *env = DefaultEnv::GetEnv();
2287
2288 //--------------------------------------------------------------------------
2289 // Compute the login cgi
2290 //--------------------------------------------------------------------------
2291 int timeZone = XrdSysTimer::TimeZone();
2292 char *hostName = XrdNetUtils::MyHostName();
2293 std::string countryCode = Utils::FQDNToCC( hostName );
2294 char *cgiBuffer = new char[1024 + info->logintoken.size()];
2295 std::string appName;
2296 std::string monInfo;
2297 env->GetString( "AppName", appName );
2298 env->GetString( "MonInfo", monInfo );
2299 if( info->logintoken.empty() )
2300 {
2301 snprintf( cgiBuffer, 1024,
2302 "xrd.cc=%s&xrd.tz=%d&xrd.appname=%s&xrd.info=%s&"
2303 "xrd.hostname=%s&xrd.rn=%s", countryCode.c_str(), timeZone,
2304 appName.c_str(), monInfo.c_str(), hostName, XrdVERSION );
2305 }
2306 else
2307 {
2308 snprintf( cgiBuffer, 1024,
2309 "xrd.cc=%s&xrd.tz=%d&xrd.appname=%s&xrd.info=%s&"
2310 "xrd.hostname=%s&xrd.rn=%s&%s", countryCode.c_str(), timeZone,
2311 appName.c_str(), monInfo.c_str(), hostName, XrdVERSION, info->logintoken.c_str() );
2312 }
2313 uint16_t cgiLen = strlen( cgiBuffer );
2314 free( hostName );
2315
2316 //--------------------------------------------------------------------------
2317 // Generate the message
2318 //--------------------------------------------------------------------------
2319 Message *msg = new Message( sizeof(ClientLoginRequest) + cgiLen );
2320 ClientLoginRequest *loginReq = (ClientLoginRequest *)msg->GetBuffer();
2321
2322 loginReq->requestid = kXR_login;
2323 loginReq->pid = ::getpid();
2324 loginReq->capver[0] = (kXR_char) kXR_asyncap | (kXR_char) kXR_ver005;
2325 loginReq->dlen = cgiLen;
2327#ifdef WITH_XRDEC
2328 loginReq->ability2 = kXR_ecredir;
2329#endif
2330
2331 int multiProtocol = 0;
2332 env->GetInt( "MultiProtocol", multiProtocol );
2333 if(multiProtocol)
2334 loginReq->ability |= kXR_multipr;
2335
2336 //--------------------------------------------------------------------------
2337 // Check the IP stacks
2338 //--------------------------------------------------------------------------
2340 bool dualStack = false;
2341 bool privateIPv6 = false;
2342 bool privateIPv4 = false;
2343
2344 if( (stacks & XrdNetUtils::hasIP64) == XrdNetUtils::hasIP64 )
2345 {
2346 dualStack = true;
2347 loginReq->ability |= kXR_hasipv64;
2348 }
2349
2350 if( (stacks & XrdNetUtils::hasIPv6) && !(stacks & XrdNetUtils::hasPub6) )
2351 {
2352 privateIPv6 = true;
2353 loginReq->ability |= kXR_onlyprv6;
2354 }
2355
2356 if( (stacks & XrdNetUtils::hasIPv4) && !(stacks & XrdNetUtils::hasPub4) )
2357 {
2358 privateIPv4 = true;
2359 loginReq->ability |= kXR_onlyprv4;
2360 }
2361
2362 // The following code snippet tries to overcome the problem that this host
2363 // may still be dual-stacked but we don't know it because one of the
2364 // interfaces was not registered in DNS.
2365 //
2366 if( !dualStack && hsData->serverAddr )
2367 {if ( ( ( stacks & XrdNetUtils::hasIPv4 )
2368 && hsData->serverAddr->isIPType(XrdNetAddrInfo::IPv6))
2369 || ( ( stacks & XrdNetUtils::hasIPv6 )
2370 && hsData->serverAddr->isIPType(XrdNetAddrInfo::IPv4)))
2371 {dualStack = true;
2372 loginReq->ability |= kXR_hasipv64;
2373 }
2374 }
2375
2376 //--------------------------------------------------------------------------
2377 // Check the username
2378 //--------------------------------------------------------------------------
2379 std::string buffer( 8, 0 );
2380 if( hsData->url->GetUserName().length() )
2381 buffer = hsData->url->GetUserName();
2382 else
2383 {
2384 char *name = new char[1024];
2385 if( !XrdOucUtils::UserName( geteuid(), name, 1024 ) )
2386 buffer = name;
2387 else
2388 buffer = "_anon_";
2389 delete [] name;
2390 }
2391 buffer.resize( 8, 0 );
2392 std::copy( buffer.begin(), buffer.end(), (char*)loginReq->username );
2393
2394 msg->Append( cgiBuffer, cgiLen, 24 );
2395
2396 log->Debug( XRootDTransportMsg, "[%s] Sending out kXR_login request, "
2397 "username: %s, cgi: %s, dual-stack: %s, private IPv4: %s, "
2398 "private IPv6: %s", hsData->streamName.c_str(),
2399 loginReq->username, cgiBuffer, dualStack ? "true" : "false",
2400 privateIPv4 ? "true" : "false",
2401 privateIPv6 ? "true" : "false" );
2402
2403 delete [] cgiBuffer;
2404 MarshallRequest( msg );
2405 return msg;
2406 }
2407
2408 //----------------------------------------------------------------------------
2409 // Process the protocol response
2410 //----------------------------------------------------------------------------
2411 XRootDStatus XRootDTransport::ProcessLogInResp( HandShakeData *hsData,
2412 XRootDChannelInfo *info )
2413 {
2414 Log *log = DefaultEnv::GetLog();
2415
2416 XRootDStatus st = UnMarshallBody( hsData->in, kXR_login );
2417 if( !st.IsOK() )
2418 return st;
2419
2420 ServerResponse *rsp = (ServerResponse*)hsData->in->GetBuffer();
2421
2422 if( rsp->hdr.status != kXR_ok )
2423 {
2424 log->Error( XRootDTransportMsg, "[%s] Got invalid login response",
2425 hsData->streamName.c_str() );
2426 return XRootDStatus( stFatal, errLoginFailed, 0, "Got invalid login response." );
2427 }
2428
2429 if( !info->firstLogIn )
2430 memcpy( info->oldSessionId, info->sessionId, 16 );
2431
2432 if( rsp->hdr.dlen == 0 && info->protocolVersion <= 0x289 )
2433 {
2434 //--------------------------------------------------------------------------
2435 // This if statement is there only to support dCache inaccurate
2436 // implementation of XRoot protocol, that in some cases returns
2437 // an empty login response for protocol version <= 2.8.9.
2438 //--------------------------------------------------------------------------
2439 memset( info->sessionId, 0, 16 );
2440 log->Warning( XRootDTransportMsg,
2441 "[%s] Logged in, accepting empty login response.",
2442 hsData->streamName.c_str() );
2443 return XRootDStatus();
2444 }
2445
2446 if( rsp->hdr.dlen < 16 )
2447 return XRootDStatus( stError, errDataError, 0, "Login response too short." );
2448
2449 memcpy( info->sessionId, rsp->body.login.sessid, 16 );
2450
2451 std::string sessId = Utils::Char2Hex( rsp->body.login.sessid, 16 );
2452
2453 log->Debug( XRootDTransportMsg, "[%s] Logged in, session: %s",
2454 hsData->streamName.c_str(), sessId.c_str() );
2455
2456 //--------------------------------------------------------------------------
2457 // We have an authentication info to process
2458 //--------------------------------------------------------------------------
2459 if( rsp->hdr.dlen > 16 )
2460 {
2461 size_t len = rsp->hdr.dlen-16;
2462 info->authBuffer = new char[len+1];
2463 info->authBuffer[len] = 0;
2464 memcpy( info->authBuffer, rsp->body.login.sec, len );
2465 log->Debug( XRootDTransportMsg, "[%s] Authentication is required: %s",
2466 hsData->streamName.c_str(), info->authBuffer );
2467
2468 return XRootDStatus( stOK, suContinue );
2469 }
2470
2471 return XRootDStatus();
2472 }
2473
2474 //----------------------------------------------------------------------------
2475 // Do the authentication
2476 //----------------------------------------------------------------------------
2477 XRootDStatus XRootDTransport::DoAuthentication( HandShakeData *hsData,
2478 XRootDChannelInfo *info )
2479 {
2480 //--------------------------------------------------------------------------
2481 // Prepare
2482 //--------------------------------------------------------------------------
2483 Log *log = DefaultEnv::GetLog();
2484 XRootDStreamInfo &sInfo = info->stream[hsData->subStreamId];
2485 XrdSecCredentials *credentials = 0;
2486 std::string protocolName;
2487
2488 //--------------------------------------------------------------------------
2489 // We're doing this for the first time
2490 //--------------------------------------------------------------------------
2491 if( sInfo.status == XRootDStreamInfo::LoginSent )
2492 {
2493 log->Debug( XRootDTransportMsg, "[%s] Sending authentication data",
2494 hsData->streamName.c_str() );
2495
2496 //------------------------------------------------------------------------
2497 // Set up the authentication environment
2498 //------------------------------------------------------------------------
2499 info->authEnv = new XrdOucEnv();
2500 info->authEnv->Put( "sockname", hsData->clientName.c_str() );
2501 info->authEnv->Put( "username", hsData->url->GetUserName().c_str() );
2502 info->authEnv->Put( "password", hsData->url->GetPassword().c_str() );
2503
2504 const URL::ParamsMap &urlParams = hsData->url->GetParams();
2505 URL::ParamsMap::const_iterator it;
2506 for( it = urlParams.begin(); it != urlParams.end(); ++it )
2507 {
2508 if( it->first.compare( 0, 4, "xrd." ) == 0 ||
2509 it->first.compare( 0, 6, "xrdcl." ) == 0 )
2510 info->authEnv->Put( it->first.c_str(), it->second.c_str() );
2511 }
2512
2513 //------------------------------------------------------------------------
2514 // Initialize some other structs
2515 //------------------------------------------------------------------------
2516 size_t authBuffLen = strlen( info->authBuffer );
2517 char *pars = (char *)malloc( authBuffLen + 1 );
2518 memcpy( pars, info->authBuffer, authBuffLen );
2519 info->authParams = new XrdSecParameters( pars, authBuffLen );
2520 sInfo.status = XRootDStreamInfo::AuthSent;
2521 delete [] info->authBuffer;
2522 info->authBuffer = 0;
2523
2524 //------------------------------------------------------------------------
2525 // Find a protocol that gives us valid credentials
2526 //------------------------------------------------------------------------
2527 XRootDStatus st = GetCredentials( credentials, hsData, info );
2528 if( !st.IsOK() )
2529 {
2530 CleanUpAuthentication( info );
2531 return st;
2532 }
2533 protocolName = info->authProtocol->Entity.prot;
2534 }
2535
2536 //--------------------------------------------------------------------------
2537 // We've been here already
2538 //--------------------------------------------------------------------------
2539 else
2540 {
2541 ServerResponse *rsp = (ServerResponse*)hsData->in->GetBuffer();
2542 protocolName = info->authProtocol->Entity.prot;
2543
2544 //------------------------------------------------------------------------
2545 // We're required to send out more authentication data
2546 //------------------------------------------------------------------------
2547 if( rsp->hdr.status == kXR_authmore )
2548 {
2549 log->Debug( XRootDTransportMsg,
2550 "[%s] Sending more authentication data for %s",
2551 hsData->streamName.c_str(), protocolName.c_str() );
2552
2553 uint32_t len = rsp->hdr.dlen;
2554 char *secTokenData = (char*)malloc( len );
2555 memcpy( secTokenData, rsp->body.authmore.data, len );
2556 XrdSecParameters *secToken = new XrdSecParameters( secTokenData, len );
2557 XrdOucErrInfo ei( "", info->authEnv);
2558 credentials = info->authProtocol->getCredentials( secToken, &ei );
2559 delete secToken;
2560
2561 //----------------------------------------------------------------------
2562 // The protocol handler refuses to give us the data
2563 //----------------------------------------------------------------------
2564 if( !credentials )
2565 {
2566 log->Error( XRootDTransportMsg,
2567 "[%s] Auth protocol handler for %s refuses to give "
2568 "us more credentials %s",
2569 hsData->streamName.c_str(), protocolName.c_str(),
2570 ei.getErrText() );
2571 CleanUpAuthentication( info );
2572 return XRootDStatus( stFatal, errAuthFailed, 0, ei.getErrText() );
2573 }
2574 }
2575
2576 //------------------------------------------------------------------------
2577 // We have succeeded
2578 //------------------------------------------------------------------------
2579 else if( rsp->hdr.status == kXR_ok )
2580 {
2581 info->authProtocolName = info->authProtocol->Entity.prot;
2582
2583 //----------------------------------------------------------------------
2584 // Do we need protection?
2585 //----------------------------------------------------------------------
2586 if( info->protRespBody )
2587 {
2588 int rc = XrdSecGetProtection( info->protection, *info->authProtocol, *info->protRespBody, info->protRespSize );
2589 if( rc > 0 )
2590 {
2591 log->Debug( XRootDTransportMsg,
2592 "[%s] XrdSecProtect loaded.", hsData->streamName.c_str() );
2593 }
2594 else if( rc == 0 )
2595 {
2596 log->Debug( XRootDTransportMsg,
2597 "[%s] XrdSecProtect: no protection needed.",
2598 hsData->streamName.c_str() );
2599 }
2600 else
2601 {
2602 log->Debug( XRootDTransportMsg,
2603 "[%s] Failed to load XrdSecProtect: %s",
2604 hsData->streamName.c_str(), XrdSysE2T( -rc ) );
2605 CleanUpAuthentication( info );
2606
2607 return XRootDStatus( stError, errAuthFailed, -rc, XrdSysE2T( -rc ) );
2608 }
2609 }
2610
2611 if( !info->protection )
2612 CleanUpAuthentication( info );
2613 else
2614 pSecUnloadHandler->Register( info->authProtocolName );
2615
2616 log->Debug( XRootDTransportMsg,
2617 "[%s] Authenticated with %s.", hsData->streamName.c_str(),
2618 protocolName.c_str() );
2619
2620 //--------------------------------------------------------------------
2621 // Clear the SSL error queue of the calling thread, as there might be
2622 // some leftover from the authentication!
2623 //--------------------------------------------------------------------
2625
2626 return XRootDStatus();
2627 }
2628 //------------------------------------------------------------------------
2629 // Failure
2630 //------------------------------------------------------------------------
2631 else if( rsp->hdr.status == kXR_error )
2632 {
2633 char *errmsg = new char[rsp->hdr.dlen-3]; errmsg[rsp->hdr.dlen-4] = 0;
2634 memcpy( errmsg, rsp->body.error.errmsg, rsp->hdr.dlen-4 );
2635 log->Error( XRootDTransportMsg,
2636 "[%s] Authentication with %s failed: %s",
2637 hsData->streamName.c_str(), protocolName.c_str(),
2638 errmsg );
2639 delete [] errmsg;
2640
2641 info->authProtocol->Delete();
2642 info->authProtocol = 0;
2643
2644 //----------------------------------------------------------------------
2645 // Find another protocol that gives us valid credentials
2646 //----------------------------------------------------------------------
2647 XRootDStatus st = GetCredentials( credentials, hsData, info );
2648 if( !st.IsOK() )
2649 {
2650 CleanUpAuthentication( info );
2651 return st;
2652 }
2653 protocolName = info->authProtocol->Entity.prot;
2654 }
2655 //------------------------------------------------------------------------
2656 // God knows what
2657 //------------------------------------------------------------------------
2658 else
2659 {
2660 info->authProtocolName = info->authProtocol->Entity.prot;
2661 CleanUpAuthentication( info );
2662
2663 log->Error( XRootDTransportMsg,
2664 "[%s] Authentication with %s failed: unexpected answer",
2665 hsData->streamName.c_str(), protocolName.c_str() );
2666 return XRootDStatus( stFatal, errAuthFailed, 0, "Authentication failed: unexpected answer." );
2667 }
2668 }
2669
2670 //--------------------------------------------------------------------------
2671 // Generate the client request
2672 //--------------------------------------------------------------------------
2673 Message *msg = new Message( sizeof(ClientAuthRequest)+credentials->size );
2674 msg->Zero();
2675 ClientRequest *req = (ClientRequest*)msg->GetBuffer();
2676 char *reqBuffer = msg->GetBuffer(sizeof(ClientAuthRequest));
2677
2678 req->header.requestid = kXR_auth;
2679 req->auth.dlen = credentials->size;
2680 memcpy( req->auth.credtype, protocolName.c_str(),
2681 protocolName.length() > 4 ? 4 : protocolName.length() );
2682
2683 memcpy( reqBuffer, credentials->buffer, credentials->size );
2684 hsData->out = msg;
2685 MarshallRequest( msg );
2686 delete credentials;
2687
2688 //------------------------------------------------------------------------
2689 // Clear the SSL error queue of the calling thread, as there might be
2690 // some leftover from the authentication!
2691 //------------------------------------------------------------------------
2693
2694 return XRootDStatus( stOK, suContinue );
2695 }
2696
2697 //------------------------------------------------------------------------
2698 // Get the initial credentials using one of the protocols
2699 //------------------------------------------------------------------------
2700 XRootDStatus XRootDTransport::GetCredentials( XrdSecCredentials *&credentials,
2701 HandShakeData *hsData,
2702 XRootDChannelInfo *info )
2703 {
2704 //--------------------------------------------------------------------------
2705 // Set up the auth handler
2706 //--------------------------------------------------------------------------
2707 Log *log = DefaultEnv::GetLog();
2708 XrdOucErrInfo ei( "", info->authEnv);
2709 XrdSecGetProt_t authHandler = GetAuthHandler();
2710 if( !authHandler )
2711 return XRootDStatus( stFatal, errAuthFailed, 0, "Could not load authentication handler." );
2712
2713 //--------------------------------------------------------------------------
2714 // Retrieve secuid and secgid, if available. These will override the fsuid
2715 // and fsgid of the current thread reading the credentials to prevent
2716 // security holes in case this process is running with elevated permissions.
2717 //--------------------------------------------------------------------------
2718 char *secuidc = (ei.getEnv()) ? ei.getEnv()->Get("xrdcl.secuid") : 0;
2719 char *secgidc = (ei.getEnv()) ? ei.getEnv()->Get("xrdcl.secgid") : 0;
2720
2721 int secuid = -1;
2722 int secgid = -1;
2723
2724 if(secuidc) secuid = atoi(secuidc);
2725 if(secgidc) secgid = atoi(secgidc);
2726
2727#ifdef __linux__
2728 ScopedFsUidSetter uidSetter(secuid, secgid, hsData->streamName);
2729 if(!uidSetter.IsOk()) {
2730 log->Error( XRootDTransportMsg, "[%s] Error while setting (fsuid, fsgid) to (%d, %d)",
2731 hsData->streamName.c_str(), secuid, secgid );
2732 return XRootDStatus( stFatal, errAuthFailed, 0, "Error while setting (fsuid, fsgid)." );
2733 }
2734#else
2735 if(secuid >= 0 || secgid >= 0) {
2736 log->Error( XRootDTransportMsg, "[%s] xrdcl.secuid and xrdcl.secgid only supported on Linux.",
2737 hsData->streamName.c_str() );
2738 return XRootDStatus( stFatal, errAuthFailed, 0, "xrdcl.secuid and xrdcl.secgid"
2739 " only supported on Linux" );
2740 }
2741#endif
2742
2743 //--------------------------------------------------------------------------
2744 // Loop over the possible protocols to find one that gives us valid
2745 // credentials
2746 //--------------------------------------------------------------------------
2747 XrdNetAddr &srvAddrInfo = *const_cast<XrdNetAddr *>(hsData->serverAddr);
2748 srvAddrInfo.SetTLS( info->encrypted );
2749 while(1)
2750 {
2751 //------------------------------------------------------------------------
2752 // Get the protocol
2753 //------------------------------------------------------------------------
2754 info->authProtocol = (*authHandler)( hsData->url->GetHostName().c_str(),
2755 srvAddrInfo,
2756 *info->authParams,
2757 &ei );
2758 if( !info->authProtocol )
2759 {
2760 log->Error( XRootDTransportMsg, "[%s] No protocols left to try",
2761 hsData->streamName.c_str() );
2762 return XRootDStatus( stFatal, errAuthFailed, 0, "No protocols left to try" );
2763 }
2764
2765 std::string protocolName = info->authProtocol->Entity.prot;
2766 log->Debug( XRootDTransportMsg, "[%s] Trying to authenticate using %s",
2767 hsData->streamName.c_str(), protocolName.c_str() );
2768
2769 //------------------------------------------------------------------------
2770 // Get the credentials from the current protocol
2771 //------------------------------------------------------------------------
2772 credentials = info->authProtocol->getCredentials( 0, &ei );
2773 if( !credentials )
2774 {
2775 log->Debug( XRootDTransportMsg,
2776 "[%s] Cannot get credentials for protocol %s: %s",
2777 hsData->streamName.c_str(), protocolName.c_str(),
2778 ei.getErrText() );
2779 info->authProtocol->Delete();
2780 continue;
2781 }
2782 return XRootDStatus( stOK, suContinue );
2783 }
2784 }
2785
2786 //------------------------------------------------------------------------
2787 // Clean up the data structures created for the authentication process
2788 //------------------------------------------------------------------------
2789 Status XRootDTransport::CleanUpAuthentication( XRootDChannelInfo *info )
2790 {
2791 if( info->authProtocol )
2792 info->authProtocol->Delete();
2793 delete info->authParams;
2794 delete info->authEnv;
2795 info->authProtocol = 0;
2796 info->authParams = 0;
2797 info->authEnv = 0;
2799 return Status();
2800 }
2801
2802 //------------------------------------------------------------------------
2803 // Clean up the data structures created for the protection purposes
2804 //------------------------------------------------------------------------
2805 Status XRootDTransport::CleanUpProtection( XRootDChannelInfo *info )
2806 {
2807 XrdSysRWLockHelper scope( pSecUnloadHandler->lock );
2808 if( pSecUnloadHandler->unloaded ) return Status( stError, errInvalidOp );
2809
2810 if( info->protection )
2811 {
2812 info->protection->Delete();
2813 info->protection = 0;
2814
2815 CleanUpAuthentication( info );
2816 }
2817
2818 if( info->protRespBody )
2819 {
2820 delete info->protRespBody;
2821 info->protRespBody = 0;
2822 info->protRespSize = 0;
2823 }
2824
2825 return Status();
2826 }
2827
2828 //----------------------------------------------------------------------------
2829 // Get the authentication function handle
2830 //----------------------------------------------------------------------------
2831 XrdSecGetProt_t XRootDTransport::GetAuthHandler()
2832 {
2833 Log *log = DefaultEnv::GetLog();
2834 char errorBuff[1024];
2835
2836 // the static constructor is invoked only once and it is guaranteed that this
2837 // is thread safe
2838 static std::atomic<XrdSecGetProt_t> authHandler( XrdSecLoadSecFactory( errorBuff, 1024 ) );
2839 auto ret = authHandler.load( std::memory_order_relaxed );
2840 if( ret ) return ret;
2841
2842 // if we are here it means we failed to load the security library for the
2843 // first time and we hope the environment changed
2844
2845 // obtain a lock
2846 static XrdSysMutex mtx;
2847 XrdSysMutexHelper lck( mtx );
2848 // check if in the meanwhile some else didn't load the library
2849 ret = authHandler.load( std::memory_order_relaxed );
2850 if( ret ) return ret;
2851
2852 // load the library
2853 ret = XrdSecLoadSecFactory( errorBuff, 1024 );
2854 authHandler.store( ret, std::memory_order_relaxed );
2855 // if we failed report an error
2856 if( !ret )
2857 {
2858 log->Error( XRootDTransportMsg,
2859 "Unable to get the security framework: %s", errorBuff );
2860 return 0;
2861 }
2862 return ret;
2863 }
2864
2865 //----------------------------------------------------------------------------
2866 // Generate the end session message
2867 //----------------------------------------------------------------------------
2868 Message *XRootDTransport::GenerateEndSession( HandShakeData *hsData,
2869 XRootDChannelInfo *info )
2870 {
2871 Log *log = DefaultEnv::GetLog();
2872
2873 //--------------------------------------------------------------------------
2874 // Generate the message
2875 //--------------------------------------------------------------------------
2876 Message *msg = new Message( sizeof(ClientEndsessRequest) );
2877 ClientEndsessRequest *endsessReq = (ClientEndsessRequest *)msg->GetBuffer();
2878
2879 endsessReq->requestid = kXR_endsess;
2880 memcpy( endsessReq->sessid, info->oldSessionId, 16 );
2881 std::string sessId = Utils::Char2Hex( endsessReq->sessid, 16 );
2882
2883 log->Debug( XRootDTransportMsg, "[%s] Sending out kXR_endsess for session:"
2884 " %s", hsData->streamName.c_str(), sessId.c_str() );
2885
2886 MarshallRequest( msg );
2887
2888 Message *sign = 0;
2889 GetSignature( msg, sign, info );
2890 if( sign )
2891 {
2892 //------------------------------------------------------------------------
2893 // Now place both the signature and the request in a single buffer
2894 //------------------------------------------------------------------------
2895 uint32_t size = sign->GetSize();
2896 sign->ReAllocate( size + msg->GetSize() );
2897 char* buffer = sign->GetBuffer( size );
2898 memcpy( buffer, msg->GetBuffer(), msg->GetSize() );
2899 msg->Grab( sign->GetBuffer(), sign->GetSize() );
2900 }
2901
2902 return msg;
2903 }
2904
2905 //----------------------------------------------------------------------------
2906 // Process the protocol response
2907 //----------------------------------------------------------------------------
2908 Status XRootDTransport::ProcessEndSessionResp( HandShakeData *hsData,
2909 XRootDChannelInfo *info )
2910 {
2911 Log *log = DefaultEnv::GetLog();
2912
2913 Status st = UnMarshallBody( hsData->in, kXR_endsess );
2914 if( !st.IsOK() )
2915 return st;
2916
2917 ServerResponse *rsp = (ServerResponse*)hsData->in->GetBuffer();
2918
2919 // If we're good, we're good!
2920 if( rsp->hdr.status == kXR_ok )
2921 return Status();
2922
2923 // we ignore not found errors as such an error means the connection
2924 // has been already terminated
2925 if( rsp->hdr.status == kXR_error && rsp->body.error.errnum == kXR_NotFound )
2926 return Status();
2927
2928 // other errors
2929 if( rsp->hdr.status == kXR_error )
2930 {
2931 std::string errorMsg( rsp->body.error.errmsg, rsp->hdr.dlen - 4 );
2932 log->Error( XRootDTransportMsg, "[%s] Got error response to "
2933 "kXR_endsess: %s", hsData->streamName.c_str(),
2934 errorMsg.c_str() );
2935 return Status( stFatal, errHandShakeFailed );
2936 }
2937
2938 // Wait Response.
2939 if( rsp->hdr.status == kXR_wait )
2940 {
2941 std::string msg( rsp->body.wait.infomsg, rsp->hdr.dlen - 4 );
2942 log->Info( XRootDTransportMsg, "[%s] Got wait response to "
2943 "kXR_endsess: %s", hsData->streamName.c_str(),
2944 msg.c_str() );
2945 hsData->out = GenerateEndSession( hsData, info );
2946 return Status( stOK, suRetry );
2947 }
2948
2949 // Any other response is protocol violation
2950 return Status( stError, errDataError );
2951 }
2952
2953 //----------------------------------------------------------------------------
2954 // Get a string representation of the server flags
2955 //----------------------------------------------------------------------------
2956 std::string XRootDTransport::ServerFlagsToStr( uint32_t flags )
2957 {
2958 std::string repr = "type: ";
2959 if( flags & kXR_isManager )
2960 repr += "manager ";
2961
2962 else if( flags & kXR_isServer )
2963 repr += "server ";
2964
2965 repr += "[";
2966
2967 if( flags & kXR_attrMeta )
2968 repr += "meta ";
2969
2970 else if( flags & kXR_attrCache )
2971 repr += "cache ";
2972
2973 else if( flags & kXR_attrProxy )
2974 repr += "proxy ";
2975
2976 else if( flags & kXR_attrSuper )
2977 repr += "super ";
2978
2979 else
2980 repr += " ";
2981
2982 repr.erase( repr.length()-1, 1 );
2983
2984 repr += "]";
2985 return repr;
2986 }
2987}
2988
2989namespace
2990{
2991 // Extract file name from a request
2992 //----------------------------------------------------------------------------
2993 char *GetDataAsString( char *msg )
2994 {
2996 char *fn = new char[req->dlen+1];
2997 memcpy( fn, msg + 24, req->dlen );
2998 fn[req->dlen] = 0;
2999 return fn;
3000 }
3001}
3002
3003namespace XrdCl
3004{
3005 //----------------------------------------------------------------------------
3006 // Get the description of a message
3007 //----------------------------------------------------------------------------
3008 void XRootDTransport::GenerateDescription( char *msg, std::ostringstream &o )
3009 {
3010 Log *log = DefaultEnv::GetLog();
3011 if( log->GetLevel() < Log::ErrorMsg )
3012 return;
3013
3014 ClientRequestHdr *req = (ClientRequestHdr *)msg;
3015 switch( req->requestid )
3016 {
3017 //------------------------------------------------------------------------
3018 // kXR_open
3019 //------------------------------------------------------------------------
3020 case kXR_open:
3021 {
3022 ClientOpenRequest *sreq = (ClientOpenRequest *)msg;
3023 o << "kXR_open (";
3024 char *fn = GetDataAsString( msg );
3025 o << "file: " << fn << ", ";
3026 delete [] fn;
3027 o << "mode: 0" << std::setbase(8) << sreq->mode << ", ";
3028 o << std::setbase(10);
3029 o << "flags: ";
3030 if( sreq->options == 0 )
3031 o << "none ";
3032 else
3033 {
3034 if( sreq->options & kXR_compress )
3035 o << "kXR_compress ";
3036 if( sreq->options & kXR_delete )
3037 o << "kXR_delete ";
3038 if( sreq->options & kXR_force )
3039 o << "kXR_force ";
3040 if( sreq->options & kXR_mkpath )
3041 o << "kXR_mkpath ";
3042 if( sreq->options & kXR_new )
3043 o << "kXR_new ";
3044 if( sreq->options & kXR_nowait )
3045 o << "kXR_nowait ";
3046 if( sreq->options & kXR_open_apnd )
3047 o << "kXR_open_apnd ";
3048 if( sreq->options & kXR_open_read )
3049 o << "kXR_open_read ";
3050 if( sreq->options & kXR_open_updt )
3051 o << "kXR_open_updt ";
3052 if( sreq->options & kXR_open_wrto )
3053 o << "kXR_open_wrto ";
3054 if( sreq->options & kXR_posc )
3055 o << "kXR_posc ";
3056 if( sreq->options & kXR_prefname )
3057 o << "kXR_prefname ";
3058 if( sreq->options & kXR_refresh )
3059 o << "kXR_refresh ";
3060 if( sreq->options & kXR_4dirlist )
3061 o << "kXR_4dirlist ";
3062 if( sreq->options & kXR_replica )
3063 o << "kXR_replica ";
3064 if( sreq->options & kXR_seqio )
3065 o << "kXR_seqio ";
3066 if( sreq->options & kXR_async )
3067 o << "kXR_async ";
3068 if( sreq->options & kXR_retstat )
3069 o << "kXR_retstat ";
3070 }
3071 o << "flagt: ";
3072 if( sreq->optiont == 0 )
3073 o << "none ";
3074 else
3075 {
3076 if( sreq->optiont & kXR_dup )
3077 o << "kXR_dup ";
3078 if( sreq->options & kXR_samefs )
3079 o << "kXR_samefs ";
3080 }
3081 o << "fhtemplt: " << FileHandleToStr( sreq->fhtemplt );
3082 o << ")";
3083 break;
3084 }
3085
3086 //------------------------------------------------------------------------
3087 // kXR_clone
3088 //------------------------------------------------------------------------
3089 case kXR_clone:
3090 {
3092 XrdProto::clone_list *dataChunk = (XrdProto::clone_list*)(msg + 24 );
3093 o << "kXR_clone ( ";
3094 o << "handle: " << FileHandleToStr( sreq->fhandle );
3095 o << std::setbase(10);
3096 o << " list [ ";
3097 for( size_t i = 0; i < req->dlen/sizeof(XrdProto::clone_list); ++i )
3098 {
3099 o << "(src_handle: ";
3100 o << FileHandleToStr( dataChunk[i].srcFH );
3101 o << ", ";
3102 o << std::setbase(10);
3103 o << "src_offset: " << dataChunk[i].srcOffs;
3104 o << ", src_length: " << dataChunk[i].srcLen;
3105 o << ", dst_offset: " << dataChunk[i].dstOffs << "); ";
3106 }
3107
3108 o << " ] )";
3109 break;
3110 }
3111
3112 //------------------------------------------------------------------------
3113 // kXR_close
3114 //------------------------------------------------------------------------
3115 case kXR_close:
3116 {
3118 o << "kXR_close (";
3119 o << "handle: " << FileHandleToStr( sreq->fhandle );
3120 o << ")";
3121 break;
3122 }
3123
3124 //------------------------------------------------------------------------
3125 // kXR_stat
3126 //------------------------------------------------------------------------
3127 case kXR_stat:
3128 {
3129 ClientStatRequest *sreq = (ClientStatRequest *)msg;
3130 o << "kXR_stat (";
3131 if( sreq->dlen )
3132 {
3133 char *fn = GetDataAsString( msg );;
3134 o << "path: " << fn << ", ";
3135 delete [] fn;
3136 }
3137 else
3138 {
3139 o << "handle: " << FileHandleToStr( sreq->fhandle );
3140 o << ", ";
3141 }
3142 o << "flags: ";
3143 if( sreq->options == 0 )
3144 o << "none";
3145 else
3146 {
3147 if( sreq->options & kXR_vfs )
3148 o << "kXR_vfs";
3149 }
3150 o << ")";
3151 break;
3152 }
3153
3154 //------------------------------------------------------------------------
3155 // kXR_read
3156 //------------------------------------------------------------------------
3157 case kXR_read:
3158 {
3159 ClientReadRequest *sreq = (ClientReadRequest *)msg;
3160 o << "kXR_read (";
3161 o << "handle: " << FileHandleToStr( sreq->fhandle );
3162 o << std::setbase(10);
3163 o << ", ";
3164 o << "offset: " << sreq->offset << ", ";
3165 o << "size: " << sreq->rlen << ")";
3166 break;
3167 }
3168
3169 //------------------------------------------------------------------------
3170 // kXR_pgread
3171 //------------------------------------------------------------------------
3172 case kXR_pgread:
3173 {
3175 o << "kXR_pgread (";
3176 o << "handle: " << FileHandleToStr( sreq->fhandle );
3177 o << std::setbase(10);
3178 o << ", ";
3179 o << "offset: " << sreq->offset << ", ";
3180 o << "size: " << sreq->rlen << ")";
3181 break;
3182 }
3183
3184 //------------------------------------------------------------------------
3185 // kXR_write
3186 //------------------------------------------------------------------------
3187 case kXR_write:
3188 {
3190 o << "kXR_write (";
3191 o << "handle: " << FileHandleToStr( sreq->fhandle );
3192 o << std::setbase(10);
3193 o << ", ";
3194 o << "offset: " << sreq->offset << ", ";
3195 o << "size: " << sreq->dlen << ")";
3196 break;
3197 }
3198
3199 //------------------------------------------------------------------------
3200 // kXR_pgwrite
3201 //------------------------------------------------------------------------
3202 case kXR_pgwrite:
3203 {
3205 o << "kXR_pgwrite (";
3206 o << "handle: " << FileHandleToStr( sreq->fhandle );
3207 o << std::setbase(10);
3208 o << ", ";
3209 o << "offset: " << sreq->offset << ", ";
3210 o << "size: " << sreq->dlen << ")";
3211 break;
3212 }
3213
3214 //------------------------------------------------------------------------
3215 // kXR_fattr
3216 //------------------------------------------------------------------------
3217 case kXR_fattr:
3218 {
3220 int nattr = sreq->numattr;
3221 int options = sreq->options;
3222 o << "kXR_fattr";
3223 switch (sreq->subcode) {
3224 case kXR_fattrGet:
3225 o << "Get";
3226 break;
3227 case kXR_fattrSet:
3228 o << "Set";
3229 break;
3230 case kXR_fattrList:
3231 o << "List";
3232 break;
3233 case kXR_fattrDel:
3234 o << "Delete";
3235 break;
3236 default:
3237 o << " unknown subcode: " << sreq->subcode;
3238 break;
3239 }
3240 o << " (handle: " << FileHandleToStr( sreq->fhandle );
3241 o << std::setbase(10);
3242 if (nattr)
3243 o << ", numattr: " << nattr;
3244 if (options) {
3245 o << ", options: ";
3246 if (options & 0x01)
3247 o << "new";
3248 if (options & 0x10)
3249 o << "list values";
3250 }
3251 o << ", total size: " << req->dlen << ")";
3252 break;
3253 }
3254
3255 //------------------------------------------------------------------------
3256 // kXR_sync
3257 //------------------------------------------------------------------------
3258 case kXR_sync:
3259 {
3260 ClientSyncRequest *sreq = (ClientSyncRequest *)msg;
3261 o << "kXR_sync (";
3262 o << "handle: " << FileHandleToStr( sreq->fhandle );
3263 o << ")";
3264 break;
3265 }
3266
3267 //------------------------------------------------------------------------
3268 // kXR_truncate
3269 //------------------------------------------------------------------------
3270 case kXR_truncate:
3271 {
3273 o << "kXR_truncate (";
3274 if( !sreq->dlen )
3275 o << "handle: " << FileHandleToStr( sreq->fhandle );
3276 else
3277 {
3278 char *fn = GetDataAsString( msg );
3279 o << "file: " << fn;
3280 delete [] fn;
3281 }
3282 o << std::setbase(10);
3283 o << ", ";
3284 o << "offset: " << sreq->offset;
3285 o << ")";
3286 break;
3287 }
3288
3289 //------------------------------------------------------------------------
3290 // kXR_readv
3291 //------------------------------------------------------------------------
3292 case kXR_readv:
3293 {
3294 unsigned char *fhandle = 0;
3295 o << "kXR_readv (";
3296
3297 o << "handle: ";
3298 readahead_list *dataChunk = (readahead_list*)(msg + 24 );
3299 fhandle = dataChunk[0].fhandle;
3300 if( fhandle )
3301 o << FileHandleToStr( fhandle );
3302 else
3303 o << "unknown";
3304 o << ", ";
3305 o << std::setbase(10);
3306 o << "chunks: [";
3307 uint64_t size = 0;
3308 for( size_t i = 0; i < req->dlen/sizeof(readahead_list); ++i )
3309 {
3310 size += dataChunk[i].rlen;
3311 o << "(offset: " << dataChunk[i].offset;
3312 o << ", size: " << dataChunk[i].rlen << "); ";
3313 }
3314 o << "], ";
3315 o << "total size: " << size << ")";
3316 break;
3317 }
3318
3319 //------------------------------------------------------------------------
3320 // kXR_writev
3321 //------------------------------------------------------------------------
3322 case kXR_writev:
3323 {
3324 unsigned char *fhandle = 0;
3325 o << "kXR_writev (";
3326
3327 XrdProto::write_list *wrtList =
3328 reinterpret_cast<XrdProto::write_list*>( msg + 24 );
3329 uint64_t size = 0;
3330 uint32_t numChunks = 0;
3331 for( size_t i = 0; i < req->dlen/sizeof(XrdProto::write_list); ++i )
3332 {
3333 fhandle = wrtList[i].fhandle;
3334 size += wrtList[i].wlen;
3335 ++numChunks;
3336 }
3337 o << "handle: ";
3338 if( fhandle )
3339 o << FileHandleToStr( fhandle );
3340 else
3341 o << "unknown";
3342 o << ", ";
3343 o << std::setbase(10);
3344 o << "chunks: " << numChunks << ", ";
3345 o << "total size: " << size << ")";
3346 break;
3347 }
3348
3349 //------------------------------------------------------------------------
3350 // kXR_locate
3351 //------------------------------------------------------------------------
3352 case kXR_locate:
3353 {
3355 char *fn = GetDataAsString( msg );;
3356 o << "kXR_locate (";
3357 o << "path: " << fn << ", ";
3358 delete [] fn;
3359 o << "flags: ";
3360 if( sreq->options == 0 )
3361 o << "none";
3362 else
3363 {
3364 if( sreq->options & kXR_refresh )
3365 o << "kXR_refresh ";
3366 if( sreq->options & kXR_prefname )
3367 o << "kXR_prefname ";
3368 if( sreq->options & kXR_nowait )
3369 o << "kXR_nowait ";
3370 if( sreq->options & kXR_force )
3371 o << "kXR_force ";
3372 if( sreq->options & kXR_compress )
3373 o << "kXR_compress ";
3374 }
3375 o << ")";
3376 break;
3377 }
3378
3379 //------------------------------------------------------------------------
3380 // kXR_mv
3381 //------------------------------------------------------------------------
3382 case kXR_mv:
3383 {
3384 ClientMvRequest *sreq = (ClientMvRequest *)msg;
3385 o << "kXR_mv (";
3386 o << "source: ";
3387 o.write( msg + sizeof( ClientMvRequest ), sreq->arg1len );
3388 o << ", ";
3389 o << "destination: ";
3390 o.write( msg + sizeof( ClientMvRequest ) + sreq->arg1len + 1, sreq->dlen - sreq->arg1len - 1 );
3391 o << ")";
3392 break;
3393 }
3394
3395 //------------------------------------------------------------------------
3396 // kXR_query
3397 //------------------------------------------------------------------------
3398 case kXR_query:
3399 {
3401 o << "kXR_query (";
3402 o << "code: ";
3403 switch( sreq->infotype )
3404 {
3405 case kXR_Qconfig: o << "kXR_Qconfig"; break;
3406 case kXR_Qckscan: o << "kXR_Qckscan"; break;
3407 case kXR_Qcksum: o << "kXR_Qcksum"; break;
3408 case kXR_Qopaque: o << "kXR_Qopaque"; break;
3409 case kXR_Qopaquf: o << "kXR_Qopaquf"; break;
3410 case kXR_Qopaqug: o << "kXR_Qopaqug"; break;
3411 case kXR_QPrep: o << "kXR_QPrep"; break;
3412 case kXR_Qspace: o << "kXR_Qspace"; break;
3413 case kXR_QStats: o << "kXR_QStats"; break;
3414 case kXR_Qvisa: o << "kXR_Qvisa"; break;
3415 case kXR_Qxattr: o << "kXR_Qxattr"; break;
3416 default: o << sreq->infotype; break;
3417 }
3418 o << ", ";
3419
3420 if( sreq->infotype == kXR_Qopaqug || sreq->infotype == kXR_Qvisa )
3421 {
3422 o << "handle: " << FileHandleToStr( sreq->fhandle );
3423 o << ", ";
3424 }
3425
3426 o << "arg length: " << sreq->dlen << ")";
3427 break;
3428 }
3429
3430 //------------------------------------------------------------------------
3431 // kXR_rm
3432 //------------------------------------------------------------------------
3433 case kXR_rm:
3434 {
3435 o << "kXR_rm (";
3436 char *fn = GetDataAsString( msg );;
3437 o << "path: " << fn << ")";
3438 delete [] fn;
3439 break;
3440 }
3441
3442 //------------------------------------------------------------------------
3443 // kXR_mkdir
3444 //------------------------------------------------------------------------
3445 case kXR_mkdir:
3446 {
3448 o << "kXR_mkdir (";
3449 char *fn = GetDataAsString( msg );
3450 o << "path: " << fn << ", ";
3451 delete [] fn;
3452 o << "mode: 0" << std::setbase(8) << sreq->mode << ", ";
3453 o << std::setbase(10);
3454 o << "flags: ";
3455 if( sreq->options[0] == 0 )
3456 o << "none";
3457 else
3458 {
3459 if( sreq->options[0] & kXR_mkdirpath )
3460 o << "kXR_mkdirpath";
3461 }
3462 o << ")";
3463 break;
3464 }
3465
3466 //------------------------------------------------------------------------
3467 // kXR_rmdir
3468 //------------------------------------------------------------------------
3469 case kXR_rmdir:
3470 {
3471 o << "kXR_rmdir (";
3472 char *fn = GetDataAsString( msg );
3473 o << "path: " << fn << ")";
3474 delete [] fn;
3475 break;
3476 }
3477
3478 //------------------------------------------------------------------------
3479 // kXR_chmod
3480 //------------------------------------------------------------------------
3481 case kXR_chmod:
3482 {
3484 o << "kXR_chmod (";
3485 char *fn = GetDataAsString( msg );
3486 o << "path: " << fn << ", ";
3487 delete [] fn;
3488 o << "mode: 0" << std::setbase(8) << sreq->mode << ")";
3489 break;
3490 }
3491
3492 //------------------------------------------------------------------------
3493 // kXR_ping
3494 //------------------------------------------------------------------------
3495 case kXR_ping:
3496 {
3497 o << "kXR_ping ()";
3498 break;
3499 }
3500
3501 //------------------------------------------------------------------------
3502 // kXR_protocol
3503 //------------------------------------------------------------------------
3504 case kXR_protocol:
3505 {
3507 o << "kXR_protocol (";
3508 o << "clientpv: 0x" << std::setbase(16) << sreq->clientpv << ")";
3509 break;
3510 }
3511
3512 //------------------------------------------------------------------------
3513 // kXR_dirlist
3514 //------------------------------------------------------------------------
3515 case kXR_dirlist:
3516 {
3517 o << "kXR_dirlist (";
3518 char *fn = GetDataAsString( msg );;
3519 o << "path: " << fn << ")";
3520 delete [] fn;
3521 break;
3522 }
3523
3524 //------------------------------------------------------------------------
3525 // kXR_set
3526 //------------------------------------------------------------------------
3527 case kXR_set:
3528 {
3529 o << "kXR_set (";
3530 char *fn = GetDataAsString( msg );;
3531 o << "data: " << fn << ")";
3532 delete [] fn;
3533 break;
3534 }
3535
3536 //------------------------------------------------------------------------
3537 // kXR_prepare
3538 //------------------------------------------------------------------------
3539 case kXR_prepare:
3540 {
3542 o << "kXR_prepare (";
3543 o << "flags: ";
3544
3545 if( sreq->options == 0 )
3546 o << "none";
3547 else
3548 {
3549 if( sreq->options & kXR_stage )
3550 o << "kXR_stage ";
3551 if( sreq->options & kXR_wmode )
3552 o << "kXR_wmode ";
3553 if( sreq->options & kXR_coloc )
3554 o << "kXR_coloc ";
3555 if( sreq->options & kXR_fresh )
3556 o << "kXR_fresh ";
3557 }
3558
3559 o << ", priority: " << (int) sreq->prty << ", ";
3560
3561 char *fn = GetDataAsString( msg );
3562 char *cursor;
3563 for( cursor = fn; *cursor; ++cursor )
3564 if( *cursor == '\n' ) *cursor = ' ';
3565
3566 o << "paths: " << fn << ")";
3567 delete [] fn;
3568 break;
3569 }
3570
3571 case kXR_chkpoint:
3572 {
3574 o << "kXR_chkpoint (";
3575 o << "opcode: ";
3576 if( sreq->opcode == kXR_ckpBegin ) o << "kXR_ckpBegin)";
3577 else if( sreq->opcode == kXR_ckpCommit ) o << "kXR_ckpCommit)";
3578 else if( sreq->opcode == kXR_ckpQuery ) o << "kXR_ckpQuery)";
3579 else if( sreq->opcode == kXR_ckpRollback ) o << "kXR_ckpRollback)";
3580 else if( sreq->opcode == kXR_ckpXeq )
3581 {
3582 o << "kXR_ckpXeq) ";
3583 // In this case our request body will be one of kXR_pgwrite,
3584 // kXR_truncate, kXR_write, or kXR_writev request.
3585 GenerateDescription( msg + sizeof( ClientChkPointRequest ), o );
3586 }
3587
3588 break;
3589 }
3590
3591 //------------------------------------------------------------------------
3592 // Default
3593 //------------------------------------------------------------------------
3594 default:
3595 {
3596 o << "kXR_unknown (length: " << req->dlen << ")";
3597 break;
3598 }
3599 };
3600 }
3601
3602 //----------------------------------------------------------------------------
3603 // Get a string representation of file handle
3604 //----------------------------------------------------------------------------
3605 std::string XRootDTransport::FileHandleToStr( const unsigned char handle[4] )
3606 {
3607 std::ostringstream o;
3608 o << "0x";
3609 for( uint8_t i = 0; i < 4; ++i )
3610 {
3611 o << std::setbase(16) << std::setfill('0') << std::setw(2);
3612 o << (int)handle[i];
3613 }
3614 return o.str();
3615 }
3616}
static const int kXR_ckpRollback
Definition XProtocol.hh:217
@ kXR_NotFound
kXR_int16 arg1len
Definition XProtocol.hh:460
#define kXR_isManager
struct ClientTruncateRequest truncate
Definition XProtocol.hh:917
@ kXR_ecredir
Definition XProtocol.hh:401
#define kXR_tlsLogin
@ kXR_fattrDel
Definition XProtocol.hh:300
@ kXR_fattrSet
Definition XProtocol.hh:303
@ kXR_fattrList
Definition XProtocol.hh:302
@ kXR_fattrGet
Definition XProtocol.hh:301
#define kXR_suppgrw
kXR_char fhandle[4]
Definition XProtocol.hh:565
kXR_unt16 requestid
Definition XProtocol.hh:424
ServerResponseStatus status
kXR_char fhandle[4]
Definition XProtocol.hh:823
#define kXR_gotoTLS
#define kXR_attrMeta
struct ClientPgReadRequest pgread
Definition XProtocol.hh:903
union ServerResponse::@040373375333017131300127053271011057331004327334 body
kXR_char fhandle[4]
Definition XProtocol.hh:848
#define kXR_haveTLS
kXR_char streamid[2]
Definition XProtocol.hh:158
kXR_char fhandle[4]
Definition XProtocol.hh:812
struct ClientMkdirRequest mkdir
Definition XProtocol.hh:900
kXR_int32 dlen
Definition XProtocol.hh:461
struct ClientAuthRequest auth
Definition XProtocol.hh:888
kXR_char streamid[2]
Definition XProtocol.hh:956
kXR_char fhtemplt[4]
Definition XProtocol.hh:516
kXR_unt16 options
Definition XProtocol.hh:513
static const int kXR_ckpXeq
Definition XProtocol.hh:218
struct ClientPgWriteRequest pgwrite
Definition XProtocol.hh:904
#define kXR_attrSuper
struct ClientReadVRequest readv
Definition XProtocol.hh:910
kXR_char pathid
Definition XProtocol.hh:689
kXR_char credtype[4]
Definition XProtocol.hh:172
kXR_char username[8]
Definition XProtocol.hh:426
@ kXR_open_wrto
Definition XProtocol.hh:499
@ kXR_compress
Definition XProtocol.hh:482
@ kXR_async
Definition XProtocol.hh:488
@ kXR_delete
Definition XProtocol.hh:483
@ kXR_prefname
Definition XProtocol.hh:491
@ kXR_nowait
Definition XProtocol.hh:497
@ kXR_open_read
Definition XProtocol.hh:486
@ kXR_open_updt
Definition XProtocol.hh:487
@ kXR_mkpath
Definition XProtocol.hh:490
@ kXR_seqio
Definition XProtocol.hh:498
@ kXR_replica
Definition XProtocol.hh:495
@ kXR_posc
Definition XProtocol.hh:496
@ kXR_refresh
Definition XProtocol.hh:489
@ kXR_new
Definition XProtocol.hh:485
@ kXR_force
Definition XProtocol.hh:484
@ kXR_4dirlist
Definition XProtocol.hh:494
@ kXR_open_apnd
Definition XProtocol.hh:492
@ kXR_retstat
Definition XProtocol.hh:493
struct ClientOpenRequest open
Definition XProtocol.hh:902
@ kXR_waitresp
Definition XProtocol.hh:948
@ kXR_redirect
Definition XProtocol.hh:946
@ kXR_status
Definition XProtocol.hh:949
@ kXR_ok
Definition XProtocol.hh:941
@ kXR_authmore
Definition XProtocol.hh:944
@ kXR_attn
Definition XProtocol.hh:943
@ kXR_wait
Definition XProtocol.hh:947
@ kXR_error
Definition XProtocol.hh:945
struct ServerResponseBody_Status bdy
struct ClientRequestHdr header
Definition XProtocol.hh:887
kXR_char fhandle[4]
Definition XProtocol.hh:543
kXR_unt16 optiont
Definition XProtocol.hh:514
kXR_char fhandle[4]
Definition XProtocol.hh:681
kXR_char fhandle[4]
Definition XProtocol.hh:695
struct ClientWriteVRequest writev
Definition XProtocol.hh:919
kXR_char fhandle[4]
Definition XProtocol.hh:258
struct ClientLoginRequest login
Definition XProtocol.hh:899
kXR_unt16 requestid
Definition XProtocol.hh:159
kXR_char fhandle[4]
Definition XProtocol.hh:669
kXR_char sessid[16]
Definition XProtocol.hh:183
@ kXR_read
Definition XProtocol.hh:126
@ kXR_open
Definition XProtocol.hh:123
@ kXR_writev
Definition XProtocol.hh:144
@ kXR_clone
Definition XProtocol.hh:145
@ kXR_readv
Definition XProtocol.hh:138
@ kXR_mkdir
Definition XProtocol.hh:121
@ kXR_sync
Definition XProtocol.hh:129
@ kXR_chmod
Definition XProtocol.hh:115
@ kXR_bind
Definition XProtocol.hh:137
@ kXR_dirlist
Definition XProtocol.hh:117
@ kXR_fattr
Definition XProtocol.hh:133
@ kXR_rm
Definition XProtocol.hh:127
@ kXR_query
Definition XProtocol.hh:114
@ kXR_write
Definition XProtocol.hh:132
@ kXR_login
Definition XProtocol.hh:120
@ kXR_auth
Definition XProtocol.hh:113
@ kXR_endsess
Definition XProtocol.hh:136
@ kXR_set
Definition XProtocol.hh:131
@ kXR_rmdir
Definition XProtocol.hh:128
@ kXR_1stRequest
Definition XProtocol.hh:112
@ kXR_truncate
Definition XProtocol.hh:141
@ kXR_protocol
Definition XProtocol.hh:119
@ kXR_mv
Definition XProtocol.hh:122
@ kXR_ping
Definition XProtocol.hh:124
@ kXR_stat
Definition XProtocol.hh:130
@ kXR_pgread
Definition XProtocol.hh:143
@ kXR_chkpoint
Definition XProtocol.hh:125
@ kXR_locate
Definition XProtocol.hh:140
@ kXR_close
Definition XProtocol.hh:116
@ kXR_pgwrite
Definition XProtocol.hh:139
@ kXR_prepare
Definition XProtocol.hh:134
struct ClientChmodRequest chmod
Definition XProtocol.hh:891
#define kXR_isServer
#define kXR_attrCache
struct ClientQueryRequest query
Definition XProtocol.hh:908
struct ClientReadRequest read
Definition XProtocol.hh:909
struct ClientMvRequest mv
Definition XProtocol.hh:901
kXR_int32 rlen
Definition XProtocol.hh:696
kXR_unt16 requestid
Definition XProtocol.hh:182
kXR_char sessid[16]
Definition XProtocol.hh:289
struct ClientChkPointRequest chkpoint
Definition XProtocol.hh:890
struct ServerResponseHeader hdr
@ kXR_asyncap
Definition XProtocol.hh:408
#define kXR_attrProxy
kXR_char options[1]
Definition XProtocol.hh:446
#define kXR_PROTOCOLVERSION
Definition XProtocol.hh:70
static const int kXR_ckpCommit
Definition XProtocol.hh:215
kXR_int64 offset
Definition XProtocol.hh:697
@ kXR_vfs
Definition XProtocol.hh:799
struct ClientPrepareRequest prepare
Definition XProtocol.hh:906
@ kXR_mkdirpath
Definition XProtocol.hh:440
@ kXR_wmode
Definition XProtocol.hh:625
@ kXR_fresh
Definition XProtocol.hh:627
@ kXR_coloc
Definition XProtocol.hh:626
@ kXR_stage
Definition XProtocol.hh:624
static const int kXR_ckpQuery
Definition XProtocol.hh:216
#define kXR_tlsSess
#define kXR_DataServer
@ kXR_dup
Definition XProtocol.hh:503
@ kXR_samefs
Definition XProtocol.hh:504
struct ClientWriteRequest write
Definition XProtocol.hh:918
#define kXR_PROTTLSVERSION
Definition XProtocol.hh:72
kXR_char capver[1]
Definition XProtocol.hh:429
struct ClientProtocolRequest protocol
Definition XProtocol.hh:907
@ kXR_QPrep
Definition XProtocol.hh:650
@ kXR_Qopaqug
Definition XProtocol.hh:661
@ kXR_Qconfig
Definition XProtocol.hh:655
@ kXR_Qopaquf
Definition XProtocol.hh:660
@ kXR_Qckscan
Definition XProtocol.hh:654
@ kXR_Qxattr
Definition XProtocol.hh:652
@ kXR_Qspace
Definition XProtocol.hh:653
@ kXR_Qvisa
Definition XProtocol.hh:656
@ kXR_QStats
Definition XProtocol.hh:649
@ kXR_Qcksum
Definition XProtocol.hh:651
@ kXR_Qopaque
Definition XProtocol.hh:659
struct ClientLocateRequest locate
Definition XProtocol.hh:898
kXR_char fhandle[4]
Definition XProtocol.hh:231
@ kXR_ver005
Definition XProtocol.hh:419
#define kXR_tlsData
@ kXR_readrdok
Definition XProtocol.hh:390
@ kXR_fullurl
Definition XProtocol.hh:388
@ kXR_onlyprv4
Definition XProtocol.hh:392
@ kXR_lclfile
Definition XProtocol.hh:394
@ kXR_multipr
Definition XProtocol.hh:389
@ kXR_redirflags
Definition XProtocol.hh:395
@ kXR_hasipv64
Definition XProtocol.hh:391
@ kXR_onlyprv6
Definition XProtocol.hh:393
ServerResponseHeader hdr
struct ClientCloneRequest clone
Definition XProtocol.hh:892
static const int kXR_ckpBegin
Definition XProtocol.hh:214
long long kXR_int64
Definition XPtypes.hh:98
unsigned char kXR_char
Definition XPtypes.hh:65
XrdVERSIONINFOREF(XrdCl)
XrdSecBuffer XrdSecParameters
XrdSecProtocol *(* XrdSecGetProt_t)(const char *hostname, XrdNetAddrInfo &endPoint, XrdSecParameters &sectoken, XrdOucErrInfo *einfo)
Typedef to simplify the encoding of methods returning XrdSecProtocol.
XrdSecBuffer XrdSecCredentials
XrdSecGetProt_t XrdSecLoadSecFactory(char *eBuff, int eBlen, const char *seclib)
int XrdSecGetProtection(XrdSecProtect *&protP, XrdSecProtocol &aprot, ServerResponseBody_Protocol &resp, unsigned int resplen)
#define NEED2SECURE(protP)
This class implements the XRootD protocol security protection.
const char * XrdSysE2T(int errcode)
Definition XrdSysE2T.cc:104
void Set(Type object, bool own=true)
void Get(Type &object)
Retrieve the object being held.
void AdvanceCursor(uint32_t delta)
Advance the cursor.
void Grab(char *buffer, uint32_t size)
Grab a buffer allocated outside.
void Zero()
Zero.
char * GetBufferAtCursor()
Get the buffer pointer at the append cursor.
void ReAllocate(uint32_t size)
Reallocate the buffer to a new location of a given size.
void Allocate(uint32_t size)
Allocate the buffer.
const char * GetBuffer(uint32_t offset=0) const
Get the message buffer.
uint32_t GetCursor() const
Get append cursor.
uint32_t GetSize() const
Get the size of the message.
static TransportManager * GetTransportManager()
Get transport manager.
static Log * GetLog()
Get default log.
static Env * GetEnv()
Get default client environment.
bool PutInt(const std::string &key, int value)
Definition XrdClEnv.cc:136
bool GetInt(const std::string &key, int &value)
Definition XrdClEnv.cc:115
Handle diagnostics.
Definition XrdClLog.hh:101
@ ErrorMsg
report errors
Definition XrdClLog.hh:109
void Error(uint64_t topic, const char *format,...)
Report an error.
Definition XrdClLog.cc:231
LogLevel GetLevel() const
Get the log level.
Definition XrdClLog.hh:258
void Dump(uint64_t topic, const char *format,...)
Print a dump message.
Definition XrdClLog.cc:299
void Debug(uint64_t topic, const char *format,...)
Print a debug message.
Definition XrdClLog.cc:282
The message representation used throughout the system.
void SetIsMarshalled(bool isMarshalled)
Set the marshalling status.
bool IsMarshalled() const
Check if the message is marshalled.
static SIDMgrPool & Instance()
std::shared_ptr< SIDManager > GetSIDMgr(const URL &url)
A network socket.
virtual XRootDStatus Read(char *buffer, size_t size, int &bytesRead)
static void ClearErrorQueue()
Clear the error queue for the calling thread.
Definition XrdClTls.cc:422
Perform the handshake and the authentication for each physical stream.
@ RequestClose
Send a close request.
virtual void WaitBeforeExit()=0
Wait before exit.
Manage transport handler objects.
TransportHandler * GetHandler(const std::string &protocol)
Get a transport handler object for a given protocol.
URL representation.
Definition XrdClURL.hh:31
std::string GetChannelId() const
Definition XrdClURL.cc:512
std::map< std::string, std::string > ParamsMap
Definition XrdClURL.hh:33
bool IsSecure() const
Does the protocol indicate encryption.
Definition XrdClURL.cc:482
bool IsTPC() const
Is the URL used in TPC context.
Definition XrdClURL.cc:490
std::string GetLoginToken() const
Get the login token if present in the opaque info.
Definition XrdClURL.cc:367
static std::string TimeToString(time_t timestamp)
Convert timestamp to a string.
static std::string FQDNToCC(const std::string &fqdn)
Convert the fully qualified host name to country code.
static std::string Char2Hex(uint8_t *array, uint16_t size)
Print a char array as hex.
static void splitString(Container &result, const std::string &input, const std::string &delimiter)
Split a string.
Definition XrdClUtils.hh:56
const std::string & GetErrorMessage() const
Get error message.
static uint16_t NbConnectedStrm(AnyObject &channelData)
Number of currently connected data streams.
virtual bool IsStreamTTLElapsed(time_t time, AnyObject &channelData)
Check if the stream should be disconnected.
virtual void Disconnect(AnyObject &channelData, uint16_t subStreamId)
The stream has been disconnected, do the cleanups.
virtual uint32_t MessageReceived(Message &msg, uint16_t subStream, AnyObject &channelData)
Check if the message invokes a stream action.
virtual void WaitBeforeExit()
Wait until the program can safely exit.
static XRootDStatus UnMarshallBody(Message *msg, uint16_t reqType)
Unmarshall the body of the incoming message.
virtual XRootDStatus GetBody(Message &message, Socket *socket)
virtual XRootDStatus GetHeader(Message &message, Socket *socket)
virtual uint16_t SubStreamNumber(AnyObject &channelData)
Return a number of substreams per stream that should be created.
virtual void FinalizeChannel(AnyObject &channelData)
Finalize channel.
virtual bool HandShakeDone(HandShakeData *handShakeData, AnyObject &channelData)
virtual Status GetSignature(Message *toSign, Message *&sign, AnyObject &channelData)
Get signature for given message.
virtual void MessageSent(Message *msg, uint16_t subStream, uint32_t bytesSent, AnyObject &channelData)
Notify the transport about a message having been sent.
virtual XRootDStatus HandShake(HandShakeData *handShakeData, AnyObject &channelData)
HandShake.
virtual XRootDStatus GetMore(Message &message, Socket *socket)
static void GenerateDescription(char *msg, std::ostringstream &o)
Get the description of a message.
static XRootDStatus UnMarshallRequest(Message *msg)
static XRootDStatus UnMarchalStatusMore(Message &msg)
Unmarshall the correction-segment of the status response for pgwrite.
static void LogErrorResponse(const Message &msg)
Log server error response.
virtual void DecFileInstCnt(AnyObject &channelData)
Decrement file object instance count bound to this channel.
virtual PathID Multiplex(Message *msg, AnyObject &channelData, PathID *hint=0)
virtual void InitializeChannel(const URL &url, AnyObject &channelData)
Initialize channel.
virtual Status Query(uint16_t query, AnyObject &result, AnyObject &channelData)
Query the channel.
static void UnMarshallHeader(Message &msg)
Unmarshall the header incoming message.
static XRootDStatus UnMarshalStatusBody(Message &msg, uint16_t reqType)
Unmarshall the body of the status response.
static XRootDStatus MarshallRequest(Message *msg)
Marshal the outgoing message.
virtual URL GetBindPreference(const URL &url, AnyObject &channelData)
Get bind preference for the next data stream.
virtual PathID MultiplexSubStream(Message *msg, AnyObject &channelData, PathID *hint=0)
virtual bool NeedEncryption(HandShakeData *handShakeData, AnyObject &channelData)
virtual Status IsStreamBroken(time_t inactiveTime, AnyObject &channelData)
void SetTLS(bool val)
static char * MyHostName(const char *eName="*unknown*", const char **eText=0)
static NetProt NetConfig(NetType netquery=qryINET, const char **eText=0)
static uint32_t Calc32C(const void *data, size_t count, uint32_t prevcs=0)
Definition XrdOucCRC.cc:190
static int UserName(uid_t uID, char *uName, int uNsz)
virtual int Secure(SecurityRequest *&newreq, ClientRequest &thereq, const char *thedata)
static int TimeZone()
const uint16_t suRetry
const uint16_t errQueryNotSupported
const int DefaultLoadBalancerTTL
const uint64_t XRootDTransportMsg
const uint16_t errTlsError
const uint16_t stFatal
Fatal error, it's still an error.
const uint16_t stError
An error occurred that could potentially be retried.
const uint16_t errLoginFailed
const int DefaultWantTlsOnNoPgrw
const uint16_t errSocketTimeout
const uint64_t XRootDMsg
const uint16_t errDataError
data is corrupted
const uint16_t errInternal
Internal error.
const uint16_t stOK
Everything went OK.
const int DefaultSubStreamsPerChannel
const uint16_t errInvalidOp
const int DefaultDataServerTTL
const uint16_t errHandShakeFailed
const int DefaultStreamTimeout
const uint16_t suAlreadyDone
const uint16_t errNotSupported
const uint16_t suDone
const uint16_t suContinue
bool InitTLS()
Definition XrdClTls.cc:96
const int DefaultTlsNoData
const int DefaultNoTlsOK
const uint16_t errAuthFailed
const uint16_t errInvalidMessage
XrdSysError Log
Definition XrdConfig.cc:113
kXR_char fhandle[4]
Definition XProtocol.hh:873
struct ServerResponseBifs_Protocol bifReqs
struct ServerResponseReqs_Protocol secReqs
kXR_char fhandle[4]
Definition XProtocol.hh:318
BindPrefSelector(std::vector< std::string > &&bindprefs)
Data structure that carries the handshake information.
std::string streamName
Name of the stream.
uint16_t subStreamId
Sub-stream id.
Message * out
Message to be sent out.
static void UnloadHandler(const std::string &trProt)
void Register(const std::string &protocol)
std::set< std::string > protocols
Procedure execution status.
uint16_t code
Error type, or additional hints on what to do.
bool IsOK() const
We're fine.
Selects less loaded stream for read operation over multiple streams.
void AdjustQueues(uint16_t size)
void MsgReceived(uint16_t substrm)
uint16_t Select(const std::vector< bool > &connected)
static const uint16_t Name
Transport name, returns const char *.
static const uint16_t Auth
Transport name, returns std::string *.
Information holder for xrootd channels.
std::vector< XRootDStreamInfo > StreamInfoVector
std::set< uint16_t > sentCloses
std::unique_ptr< StreamSelector > strmSelector
std::unique_ptr< BindPrefSelector > bindSelector
std::atomic< uint32_t > finstcnt
ServerResponseBody_Protocol * protRespBody
std::shared_ptr< SIDManager > sidManager
static const uint16_t ServerFlags
returns server flags
static const uint16_t ProtocolVersion
returns the protocol version
static const uint16_t IsEncrypted
returns true if the channel is encrypted
Information holder for XRootDStreams.
char * buffer
Pointer to the buffer.
int size
Size of the buffer or length of data in the buffer.