Permission check for ddmadmin done wrong
Motivation
----------
In some of the permission checks we do
`if _is_root(issuer) or issuer == 'ddmadmin':`
which does not work anymore since we introduced the InternalAccount for the issuers.

Modification
------------
Change to
`if _is_root(issuer) or issuer.external == 'ddmadmin':`
or add ddmadmin to the `_is_root` check


diff --git a/lib/rucio/client/downloadclient.py b/lib/rucio/client/downloadclient.py
index 8872c476a..7b916bfff 100644
--- a/lib/rucio/client/downloadclient.py
+++ b/lib/rucio/client/downloadclient.py
@@ -27,7 +27,6 @@ from __future__ import division
 import copy
 import logging
 import os
-import os.path
 import random
 import shutil
 import signal
@@ -40,7 +39,7 @@ except ImportError:
 from threading import Thread
 
 from rucio.client.client import Client
-from rucio.common.exception import (InputValidationError, NoFilesDownloaded, NotAllFilesDownloaded, RSENotFound, RucioException)
+from rucio.common.exception import (InputValidationError, NoFilesDownloaded, NotAllFilesDownloaded, RucioException)
 from rucio.common.pcache import Pcache
 from rucio.common.utils import adler32, md5, detect_client_location, generate_uuid, parse_replicas_from_string, send_trace, sizefmt, execute, parse_replicas_from_file
 from rucio.rse import rsemanager as rsemgr
@@ -471,8 +470,8 @@ class DownloadClient:
         if not sources or not len(sources):
             logger.warning('%sNo available source found for file: %s' % (log_prefix, did_str))
             item['clientState'] = 'FILE_NOT_FOUND'
-
             trace['clientState'] = 'FILE_NOT_FOUND'
+            trace['stateReason'] = 'No available sources'
             self._send_trace(trace)
             return item
 
@@ -526,8 +525,9 @@ class DownloadClient:
 
             try:
                 rse = rsemgr.get_rse_info(rse_name)
-            except RSENotFound:
-                logger.warning('%sCould not get info of RSE %s' % (log_prefix, rse_name))
+            except RucioException as error:
+                logger.warning('%sCould not get info of RSE %s: %s' % (log_prefix, rse_name, error))
+                trace['stateReason'] = str(error)
                 continue
 
             trace['remoteSite'] = rse_name
@@ -542,6 +542,7 @@ class DownloadClient:
             except Exception as error:
                 logger.warning('%sFailed to create protocol for PFN: %s' % (log_prefix, pfn))
                 logger.debug('scheme: %s, exception: %s' % (scheme, error))
+                trace['stateReason'] = str(error)
                 continue
 
             attempt = 0
@@ -563,6 +564,7 @@ class DownloadClient:
                 except Exception as error:
                     logger.debug(error)
                     trace['clientState'] = str(type(error).__name__)
+                    trace['stateReason'] = str(error)
 
                 end_time = time.time()
 
@@ -584,6 +586,7 @@ class DownloadClient:
                         logger.warning('%sChecksum validation failed for file: %s' % (log_prefix, did_str))
                         logger.debug('Local checksum: %s, Rucio checksum: %s' % (local_checksum, rucio_checksum))
                         trace['clientState'] = 'FAIL_VALIDATE'
+                        trace['stateReason'] = 'Checksum validation failed: Local checksum: %s, Rucio checksum: %s' % (local_checksum, rucio_checksum)
                 if not success:
                     logger.warning('%sDownload attempt failed. Try %s/%s' % (log_prefix, attempt, retries))
                     self._send_trace(trace)
@@ -616,6 +619,7 @@ class DownloadClient:
         trace['transferStart'] = start_time
         trace['transferEnd'] = end_time
         trace['clientState'] = 'DONE'
+        trace['stateReason'] = 'OK'
         item['clientState'] = 'DONE'
         self._send_trace(trace)
 
diff --git a/lib/rucio/client/uploadclient.py b/lib/rucio/client/uploadclient.py
index 401ea3941..7cc33dbe8 100644
--- a/lib/rucio/client/uploadclient.py
+++ b/lib/rucio/client/uploadclient.py
@@ -68,7 +68,7 @@ class UploadClient:
         self.trace['eventType'] = 'upload'
         self.trace['eventVersion'] = version.RUCIO_VERSION[0]
 
-    def upload(self, items, summary_file_path=None):
+    def upload(self, items, summary_file_path=None, traces_copy_out=None):
         """
         :param items: List of dictionaries. Each dictionary describing a file to upload. Keys:
             path                  - path of the file that will be uploaded
@@ -85,6 +85,7 @@ class UploadClient:
             transfer_timeout      - Optional: time after the upload will be aborted
             guid                  - Optional: guid of the file
         :param summary_file_path: Optional: a path where a summary in form of a json file will be stored
+        :param traces_copy_out: reference to an external list, where the traces should be uploaded
 
         :returns: 0 on success
 
@@ -140,11 +141,16 @@ class UploadClient:
             force_scheme = file.get('force_scheme')
             delete_existing = False
 
-            self.trace['scope'] = file['did_scope']
-            self.trace['datasetScope'] = file.get('dataset_scope', '')
-            self.trace['dataset'] = file.get('dataset_name', '')
-            self.trace['remoteSite'] = rse
-            self.trace['filesize'] = file['bytes']
+            trace = copy.deepcopy(self.trace)
+            # appending trace to list reference, if the reference exists
+            if traces_copy_out is not None:
+                traces_copy_out.append(trace)
+
+            trace['scope'] = file['did_scope']
+            trace['datasetScope'] = file.get('dataset_scope', '')
+            trace['dataset'] = file.get('dataset_name', '')
+            trace['remoteSite'] = rse
+            trace['filesize'] = file['bytes']
 
             file_did = {'scope': file['did_scope'], 'name': file['did_name']}
             dataset_did_str = file.get('dataset_did_str')
@@ -170,6 +176,7 @@ class UploadClient:
                     try:
                         self.client.get_did(file['did_scope'], file['did_name'])
                         logger.info('File already registered. Skipping upload.')
+                        trace['stateReason'] = 'File already exists'
                         continue
                     except DataIdentifierNotFound:
                         logger.info('File already exists on RSE. Previous left overs will be overwritten.')
@@ -177,13 +184,16 @@ class UploadClient:
             elif not is_deterministic and not no_register:
                 if rsemgr.exists(rse_settings, pfn):
                     logger.info('File already exists on RSE with given pfn. Skipping upload. Existing replica has to be removed first.')
+                    trace['stateReason'] = 'File already exists'
                     continue
                 elif rsemgr.exists(rse_settings, file_did):
                     logger.info('File already exists on RSE with different pfn. Skipping upload.')
+                    trace['stateReason'] = 'File already exists'
                     continue
             else:
                 if rsemgr.exists(rse_settings, pfn if pfn else file_did):
                     logger.info('File already exists on RSE. Skipping upload')
+                    trace['stateReason'] = 'File already exists'
                     continue
             protocols = rsemgr.get_protocols_ordered(rse_settings=rse_settings, operation='write', scheme=force_scheme)
             protocols.reverse()
@@ -204,8 +214,8 @@ class UploadClient:
                 if cur_scheme == 'https':
                     sign_service = rse_sign_service
 
-                self.trace['protocol'] = cur_scheme
-                self.trace['transferStart'] = time.time()
+                trace['protocol'] = cur_scheme
+                trace['transferStart'] = time.time()
                 try:
                     state = rsemgr.upload(rse_settings=rse_settings,
                                           lfns=lfn,
@@ -224,11 +234,11 @@ class UploadClient:
 
             if success:
                 num_succeeded += 1
-                self.trace['transferEnd'] = time.time()
-                self.trace['clientState'] = 'DONE'
+                trace['transferEnd'] = time.time()
+                trace['clientState'] = 'DONE'
                 file['state'] = 'A'
                 logger.info('Successfully uploaded file %s' % basename)
-                self._send_trace(self.trace)
+                self._send_trace(trace)
 
                 if summary_file_path:
                     summary.append(copy.deepcopy(file))
@@ -248,9 +258,9 @@ class UploadClient:
                         logger.warning('Failed to attach file to the dataset')
                         logger.debug(error)
             else:
-                self.trace['clientState'] = 'FAILED'
-                self.trace['stateReason'] = state_reason
-                self._send_trace(self.trace)
+                trace['clientState'] = 'FAILED'
+                trace['stateReason'] = state_reason
+                self._send_trace(trace)
                 logger.error('Failed to upload file %s' % basename)
 
         if summary_file_path:
diff --git a/lib/rucio/core/permission/atlas.py b/lib/rucio/core/permission/atlas.py
index 8972e2507..64b8450eb 100644
--- a/lib/rucio/core/permission/atlas.py
+++ b/lib/rucio/core/permission/atlas.py
@@ -14,7 +14,7 @@
 #
 # Authors:
 # - Vincent Garonne <vgaronne@gmail.com>, 2016
-# - Martin Barisits <martin.barisits@cern.ch>, 2016-2018
+# - Martin Barisits <martin.barisits@cern.ch>, 2016-2019
 # - Cedric Serfon <cedric.serfon@cern.ch>, 2016-2019
 # - Mario Lassnig <mario.lassnig@cern.ch>, 2018
 # - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018-2019
@@ -115,7 +115,7 @@ def has_permission(issuer, action, kwargs):
 
 
 def _is_root(issuer):
-    return issuer.external == 'root'
+    return issuer.external == 'root' or issuer.external == 'ddmadmin'
 
 
 def perm_default(issuer, kwargs):
@@ -448,7 +448,7 @@ def perm_del_rule(issuer, kwargs):
     :param kwargs: List of arguments for the action.
     :returns: True if account is allowed to call the API call, otherwise False
     """
-    if _is_root(issuer) or issuer == 'ddmadmin':
+    if _is_root(issuer):
         return True
     if get_rule(kwargs['rule_id'])['account'] == issuer:
         return True
diff --git a/lib/rucio/core/replica.py b/lib/rucio/core/replica.py
index 63154a967..6b4792393 100644
--- a/lib/rucio/core/replica.py
+++ b/lib/rucio/core/replica.py
@@ -952,9 +952,10 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
                             values()[0]
 
                         # server side root proxy handling if location is set.
+                        # supports root and http destinations
                         # cannot be pushed into protocols because we need to lookup rse attributes.
                         # ultra-conservative implementation.
-                        if domain == 'wan' and protocol.attributes['scheme'] == 'root' and client_location:
+                        if domain == 'wan' and protocol.attributes['scheme'] in ['root', 'http', 'https'] and client_location:
 
                             if 'site' in client_location and client_location['site']:
 
@@ -972,7 +973,8 @@ def _list_replicas(dataset_clause, file_clause, state_clause, show_pfns,
                                                                      default='',               # empty string to circumvent exception
                                                                      session=session)
                                     if root_proxy_internal:
-                                        pfn = 'root://' + root_proxy_internal + '//' + pfn
+                                        # don't forget to mangle gfal-style davs URL into generic https URL
+                                        pfn = 'root://' + root_proxy_internal + '//' + pfn.replace('davs://', 'https://')
 
                         # do we need to sign the URLs?
                         if sign_urls and protocol.attributes['scheme'] == 'https':
diff --git a/lib/rucio/rse/protocols/gfal.py b/lib/rucio/rse/protocols/gfal.py
index 7c2fc42bc..8af60398f 100644
--- a/lib/rucio/rse/protocols/gfal.py
+++ b/lib/rucio/rse/protocols/gfal.py
@@ -1,4 +1,4 @@
-# Copyright 2014-2018 CERN for the benefit of the ATLAS collaboration.
+# Copyright 2014-2019 CERN for the benefit of the ATLAS collaboration.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -16,7 +16,7 @@
 # - Wen Guan <wguan.icedew@gmail.com>, 2014-2016
 # - Vincent Garonne <vgaronne@gmail.com>, 2014-2018
 # - Cedric Serfon <cedric.serfon@cern.ch>, 2014-2016
-# - Mario Lassnig <mario.lassnig@cern.ch>, 2016-2017
+# - Mario Lassnig <mario.lassnig@cern.ch>, 2016-2019
 # - Tobias Wegner <twegner@cern.ch>, 2017
 # - Nicolo Magini <Nicolo.Magini@cern.ch>, 2018-2019
 # - Joaquin Bogado <jbogado@linti.unlp.edu.ar>, 2018
@@ -182,10 +182,9 @@ class Default(protocol.RSEProtocol):
         """
 
         self.__ctx = gfal2.creat_context()  # pylint: disable=no-member
-        # self.__ctx.set_opt_string("X509", "CERT", proxy)
-        # self.__ctx.set_opt_string("X509", "KEY", proxy)
         self.__ctx.set_opt_string_list("SRM PLUGIN", "TURL_PROTOCOLS", ["gsiftp", "rfio", "gsidcap", "dcap", "kdcap"])
         self.__ctx.set_opt_string("XROOTD PLUGIN", "XRD.WANTPROT", "gsi,unix")
+        self.__ctx.set_opt_boolean("XROOTD PLUGIN", "NORMALIZE_PATH", False)
 
     def get(self, path, dest, transfer_timeout=None):
         """
diff --git a/lib/rucio/transfertool/fts3.py b/lib/rucio/transfertool/fts3.py
index 1463477f5..a9c9c6047 100644
--- a/lib/rucio/transfertool/fts3.py
+++ b/lib/rucio/transfertool/fts3.py
@@ -218,7 +218,7 @@ class FTS3Transfertool(Transfertool):
             record_counter('transfertool.fts3.%s.submission.failure' % self.__extract_host(self.external_host), len(files))
 
         if not transfer_id:
-            raise('No transfer id returned by %s' % self.external_host)
+            raise TransferToolWrongAnswer('No transfer id returned by %s' % self.external_host)
         return transfer_id
 
     def cancel(self, transfer_ids, timeout=None):
['lib/rucio/rse/protocols/gfal.py', 'lib/rucio/core/permission/atlas.py', 'lib/rucio/transfertool/fts3.py', 'lib/rucio/core/replica.py', 'lib/rucio/client/uploadclient.py', 'lib/rucio/client/downloadclient.py']