diff --git a/src/peppol_py/as4_sender.py b/src/peppol_py/as4_sender.py index 697d3b2..459a204 100644 --- a/src/peppol_py/as4_sender.py +++ b/src/peppol_py/as4_sender.py @@ -29,9 +29,9 @@ def get_headers_and_body_for_posting_as4_document(document_content, document_xml return body, headers -def post_edelivery_as4_document(endpoint_url, body, headers, timeout): +def post_edelivery_as4_document(endpoint_url, body, headers, timeout, user_agent): try: - r = requests.post(endpoint_url, data=body, headers=headers, timeout=timeout) + r = requests.post(endpoint_url, data=body, headers={"User-Agent": user_agent, **headers}, timeout=timeout) except (ConnectionError, requests.exceptions.RequestException) as e: raise make_sendpeppol_error(str(e), 'server-error', temporary=True, url=endpoint_url) diff --git a/src/peppol_py/sender.py b/src/peppol_py/sender.py index d329680..6233978 100644 --- a/src/peppol_py/sender.py +++ b/src/peppol_py/sender.py @@ -120,6 +120,7 @@ def send_peppol_document( test_environment: bool=True, timeout: int=20, dryrun: bool=False, + user_agent: str="peppol-py", ) -> dict: """ Send a peppol document. Returned is a dictionary of information you need to record to later send reports to Peppol. @@ -151,6 +152,8 @@ def send_peppol_document( ``dryrun`` (bool): if specified, will prepare, get the endpoint, test document for validation errors but not send to remote endpoint. Return value will be a tuple of ``body, header, stats``. + + ``user_agent`` (str): if specified, will replace the user agent sent on HTTP connections """ document_xml = etree.fromstring(document_content) #print(etree.tostring(document_xml, pretty_print=True).decode()) @@ -176,7 +179,7 @@ def send_peppol_document( document_content, document_xml = wrap_ubl_in_peppol_standard_business_document_header(document_xml, utc_timestamp, document_type, process_type, sender_id, sender_country, receiver_id) - transport_profile, endpoint_url, receiver_cert = get_service_info_for_participant_from_smp(receiver_id, document_type, test_environment=test_environment, timeout=timeout) + transport_profile, endpoint_url, receiver_cert = get_service_info_for_participant_from_smp(receiver_id, document_type, test_environment=test_environment, timeout=timeout, user_agent=user_agent) if not endpoint_url: raise make_sendpeppol_error("Endpoint URL not found", 'not-found-in-smp') @@ -206,6 +209,6 @@ def send_peppol_document( if dryrun: return body, headers, stats - post_edelivery_as4_document(endpoint_url, body, headers, timeout=timeout) + post_edelivery_as4_document(endpoint_url, body, headers, timeout=timeout, user_agent=user_agent) return stats diff --git a/src/peppol_py/smp.py b/src/peppol_py/smp.py index 347f186..d184291 100644 --- a/src/peppol_py/smp.py +++ b/src/peppol_py/smp.py @@ -50,12 +50,12 @@ def get_smp_url_from_dns(participant_id, test_environment): return result.rstrip("/") + "/iso6523-actorid-upis::" + participant_id -def get_service_urls_for_participant_from_smp(participant_id, test_environment, timeout): +def get_service_urls_for_participant_from_smp(participant_id, test_environment, timeout, user_agent): # SML: receiver -> SMP domain (DNS) smp_url = get_smp_url_from_dns(participant_id, test_environment) try: - r = requests.get(smp_url, timeout=timeout) + r = requests.get(smp_url, timeout=timeout, headers={"User-Agent": user_agent}) r.raise_for_status() except (ConnectionError, requests.exceptions.RequestException) as e: temporary = True @@ -80,18 +80,18 @@ def get_service_urls_for_participant_from_smp(participant_id, test_environment, return service_urls -def get_service_info_for_participant_from_smp(participant_id, document_type, test_environment, timeout): +def get_service_info_for_participant_from_smp(participant_id, document_type, test_environment, timeout, user_agent): """Lookup participant in Peppol Service Metadata Publisher for information on service type. Service type is Scope.Identifier + '::' + Scope.InstanceIdentifier from the Peppol header.""" - service_urls = get_service_urls_for_participant_from_smp(participant_id, test_environment, timeout) + service_urls = get_service_urls_for_participant_from_smp(participant_id, test_environment, timeout, user_agent) service_url = next((url for url in service_urls if url.endswith(urllib.parse.quote(document_type))), None) if not service_url: raise make_sendpeppol_error("{0} not found in {1}".format(document_type, [urllib.parse.unquote(url) for url in service_urls]), 'not-found-in-smp') try: - r = requests.get(service_url, timeout=timeout) + r = requests.get(service_url, timeout=timeout, headers={"User-Agent": user_agent}) r.raise_for_status() except (ConnectionError, requests.exceptions.RequestException) as e: temporary = True