Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ pytest = "^7.1.3"
requests = "^2.28.1"
jupyter = "^1.0.0"
ipython = "^8.5.0"
security = "==1.3.1"
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This library holds security tools for protecting Python API calls.

License: MITOpen SourceMore facts



[tool.poetry.group.dev.dependencies]
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,10 +3,10 @@
Purpose: HtML parsing, using lxml
- lxml has simple syntax and faster in performance
"""
import requests
from lxml import html
from security import safe_requests

page = requests.get("https://html.com/", timeout=60)
page = safe_requests.get("https://html.com/", timeout=60)
tree = html.fromstring(page.content)

with open("html_webpage.html", mode="w", encoding="utf-8") as f:
Expand Down
5 changes: 2 additions & 3 deletions python3/12_Logging/d_sentry/a_example.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
import logging
import time

import requests
import sentry_sdk
from sentry_sdk.integrations.logging import LoggingIntegration
from security import safe_requests

# Initialize Sentry
sentry_logging = LoggingIntegration(
Expand Down Expand Up @@ -87,7 +86,7 @@

# Example 9: Tracking HTTP requests:
def get_weather_info():
response = requests.get(
response = safe_requests.get(
"https://api.openweathermap.org/data/2.5/weather?q=London&appid=YOUR_API_KEY",
timeout=60,
)
Expand Down
4 changes: 2 additions & 2 deletions python3/14_Code_Quality/05_mocking/example_3/api_access.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import requests
from security import safe_requests


def api():
response = requests.get("https://www.google.com/", timeout=60)
response = safe_requests.get("https://www.google.com/", timeout=60)
return response.status_code
4 changes: 2 additions & 2 deletions python3/14_Code_Quality/05_mocking/example_4/blog_script.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
import requests
from security import safe_requests


class Blog:
def __init__(self, name):
self.name = name

def posts(self):
response = requests.get(
response = safe_requests.get(
"https://jsonplaceholder.typicode.com/posts", timeout=60
)

Expand Down
3 changes: 2 additions & 1 deletion python3/14_Code_Quality/06_pytest/monkey_patching_test.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
import requests
from security import safe_requests


def get_number_fact(number):
url = f"http://numbersapi.com/{number}?json"
response = requests.get(url, timeout=60)
response = safe_requests.get(url, timeout=60)
json_resp = response.json()

if json_resp["found"]:
Expand Down
Original file line number Diff line number Diff line change
@@ -1,13 +1,12 @@
import re

import requests
from security import safe_requests

# get url
# url = input("Enter a URL (include `http://`): ")
url = "https://stackoverflow.com"

# connect to the url
website = requests.get(url, timeout=60)
website = safe_requests.get(url, timeout=60)

# read html
html = website.text
Expand Down
Original file line number Diff line number Diff line change
@@ -1,13 +1,12 @@
import re

import requests
from security import safe_requests

# get url
# url = input("Enter a URL (include `http://`): ")
url = "https://stackoverflow.com"

# connect to the url
website = requests.get(url, timeout=60)
website = safe_requests.get(url, timeout=60)

# read html
html = website.text
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,14 +16,13 @@
"""

import json

import requests
from security import safe_requests

# pip install -U requests


def get_data_n_write_to_file(URL):
response = requests.get(URL, timeout=60)
response = safe_requests.get(URL, timeout=60)
print(response.status_code)
if 200 <= response.status_code < 300:
# storing in a json file
Expand Down
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
# -*- coding: utf-8 -*-
from pprint import pprint

import requests
from security import safe_requests

# TO get API key, sign-up at https://api.nasa.gov/index.html#apply-for-an-api-key
NASA_API_KEY = "jdm0WrfEUq3rfZnUP8XYhvAU7QEnG7SUNY1lmiHP"
Expand All @@ -25,15 +24,15 @@
}

URL = API_ROOT + SEARCH_ENDPOINT
response = requests.get(URL, params=request_params, timeout=60)
response = safe_requests.get(URL, params=request_params, timeout=60)
# pprint(response.json())
with open("nasa_search_response.json", "w") as f:
f.write(response.text)
f.close()

# media details
URL = API_ROOT + ASSET_ENDPOINT.format(nasa_id="as11-40-5874")
response = requests.get(URL, timeout=60)
response = safe_requests.get(URL, timeout=60)
pprint(response.json())
response_data = response.json()

Expand All @@ -42,7 +41,7 @@
image_urls.append(each.get("href"))

for each_image_url in image_urls:
response_image = requests.get(each_image_url, timeout=60)
response_image = safe_requests.get(each_image_url, timeout=60)
image_name = each_image_url.split("/")[-1]
if response.headers["content-type"] == "application/json":
with open(f"{image_name}", "wb") as g:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,8 +4,7 @@
"""
import sys
import urllib

import requests
from security import safe_requests


def download_sharepoint_file(sharepoint_url, username, password, file_path):
Expand All @@ -15,7 +14,7 @@ def download_sharepoint_file(sharepoint_url, username, password, file_path):

file_path = file_path.replace(" ", "%20")

response = requests.get(
response = safe_requests.get(
file_path,
auth=HttpNtlmAuth(domain + "\\" + username, password),
stream=True,
Expand Down
Original file line number Diff line number Diff line change
@@ -1,10 +1,11 @@
import requests
from security import safe_requests

# Base URL for the HTTPBin API
base_url = "https://httpbin.org"

# Send a GET request to the /get endpoint
response = requests.get(f"{base_url}/get", timeout=60)
response = safe_requests.get(f"{base_url}/get", timeout=60)
print(response.json())

# Send a POST request to the /post endpoint with a JSON payload
Expand Down Expand Up @@ -35,42 +36,42 @@
print(response.json())

# Send a GET request to the /status/{code} endpoint to simulate an error
response = requests.get(f"{base_url}/status/404", timeout=60)
response = safe_requests.get(f"{base_url}/status/404", timeout=60)
print(response.status_code)


# Send a GET request to the /stream/{n} endpoint to stream n lines of text
response = requests.get(f"{base_url}/stream/5", stream=True, timeout=60)
response = safe_requests.get(f"{base_url}/stream/5", stream=True, timeout=60)
for line in response.iter_lines():
print(line.decode())

# Send a GET request to the /delay/{n} endpoint to simulate a delay of n seconds
response = requests.get(f"{base_url}/delay/3", timeout=60)
response = safe_requests.get(f"{base_url}/delay/3", timeout=60)
print(response.json())

# Send a GET request to the /bytes/{n} endpoint to retrieve n bytes of random data
response = requests.get(f"{base_url}/bytes/10", timeout=60)
response = safe_requests.get(f"{base_url}/bytes/10", timeout=60)
print(response.content)

# Send a GET request to the /cookies endpoint to retrieve cookies set by the server
response = requests.get(f"{base_url}/cookies", timeout=60)
response = safe_requests.get(f"{base_url}/cookies", timeout=60)
print(response.cookies)

# Send a GET request to the /cookies/set endpoint to set a cookie
response = requests.get(
response = safe_requests.get(
f"{base_url}/cookies/set", cookies={"name": "Alice"}, timeout=60
)
print(response.cookies)

# Send a GET request to the /basic-auth/{user}/{passwd} endpoint to authenticate with HTTP basic authentication
response = requests.get(
response = safe_requests.get(
f"{base_url}/basic-auth/user/passwd", auth=("user", "passwd"), timeout=60
)
print(response.json())

# Send a GET request to the /bearer endpoint to authenticate with a bearer token
headers = {"Authorization": "Bearer mytoken"}
response = requests.get(f"{base_url}/bearer", headers=headers, timeout=60)
response = safe_requests.get(f"{base_url}/bearer", headers=headers, timeout=60)
print(response.json())

# Send a POST request to the /anything endpoint to receive and return any data
Expand All @@ -89,13 +90,13 @@
print(response.content)

# Send a GET request to the /redirect/{n} endpoint to follow n redirects
response = requests.get(f"{base_url}/redirect/3", timeout=60)
response = safe_requests.get(f"{base_url}/redirect/3", timeout=60)
print(response.history)
print(response.url)

# Send a GET request to the /redirect-to endpoint to redirect to a URL
params = {"url": "https://www.google.com"}
response = requests.get(f"{base_url}/redirect-to", params=params, timeout=60)
response = safe_requests.get(f"{base_url}/redirect-to", params=params, timeout=60)
print(response.url)

# Send a POST request to the /post endpoint with a file upload
Expand All @@ -104,31 +105,31 @@
print(response.json())

# Send a GET request to the /user-agent endpoint to retrieve the user agent string
response = requests.get(f"{base_url}/user-agent", timeout=60)
response = safe_requests.get(f"{base_url}/user-agent", timeout=60)
print(response.json())

# Send a GET request to the /ip endpoint to retrieve the client's IP address
response = requests.get(f"{base_url}/ip", timeout=60)
response = safe_requests.get(f"{base_url}/ip", timeout=60)
print(response.json())

# Send a GET request to the /headers endpoint with custom headers
headers = {"X-My-Header": "123"}
response = requests.get(f"{base_url}/headers", headers=headers, timeout=60)
response = safe_requests.get(f"{base_url}/headers", headers=headers, timeout=60)
print(response.json())

# Send a GET request to the /hidden-basic-auth/{user}/{passwd} endpoint to authenticate with HTTP basic authentication
response = requests.get(
response = safe_requests.get(
f"{base_url}/hidden-basic-auth/user/passwd", auth=("user", "passwd"), timeout=60
)
print(response.json())

# Send a GET request to the /digest-auth/{qop}/{user}/{passwd}/{algorithm} endpoint to authenticate with digest authentication
response = requests.get(
response = safe_requests.get(
f"{base_url}/digest-auth/auth/user/passwd/MD5", auth=("user", "passwd"), timeout=10
)
print(response.json())

# Send a GET request to the /bearer-auth/{token} endpoint to authenticate with a bearer token
headers = {"Authorization": "Bearer mytoken"}
response = requests.get(f"{base_url}/bearer-auth/mytoken", headers=headers, timeout=60)
response = safe_requests.get(f"{base_url}/bearer-auth/mytoken", headers=headers, timeout=60)
print(response.json())
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
import json

import requests
from security import safe_requests

# r = requests.get('https://httpbin.org/stream/5', stream=True)

Expand All @@ -26,7 +25,7 @@


print("\n\n\n")
r = requests.get("https://httpbin.org/stream/5", stream=True, timeout=60)
r = safe_requests.get("https://httpbin.org/stream/5", stream=True, timeout=60)

if r.encoding is None:
r.encoding = "utf-8"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,16 +4,15 @@
https://currencylayer.com/quickstart
"""
from datetime import datetime

import requests
from security import safe_requests

ACCESS_KEY = "96b5ca6a3116caa7a9b8985fd294243e"
API_URL = "http://www.apilayer.net/api/"


def get_live_currency_quote():
URL = API_URL + "live?access_key=" + ACCESS_KEY
response = requests.get(URL, timeout=60)
response = safe_requests.get(URL, timeout=60)
# pprint(response.json())
quotes = response.json().get("quotes")
USDINR_quote = quotes.get("USDINR")
Expand All @@ -36,7 +35,7 @@ def get_live_currency_quote2(requesting_data="live"):
"format": 1,
# 'source': 'INR'
}
response = requests.get(URL, params=request_params, timeout=60).json()
response = safe_requests.get(URL, params=request_params, timeout=60).json()
if response.get("error", {}):
return response.get("error", {}).get("info")

Expand Down
Original file line number Diff line number Diff line change
@@ -1,12 +1,12 @@
import requests
from security import safe_requests


def download(url):
"""Download the given url and saves it to the current directory.

:arg url: URL of the file to be downloaded.
"""
req = requests.get(url, timeout=60)
req = safe_requests.get(url, timeout=60)
# First let us check non existing files.
if req.status_code == 404:
print("No such file found at %s" % url)
Expand Down
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
import requests
from security import safe_requests

DUCKDUCKGO_URL = "https://api.duckduckgo.com/?q={query_string}&format=json&pretty=1"

search_string = input("Enter the content to search:\n")
response = requests.get(DUCKDUCKGO_URL.format(query_string=search_string), timeout=60)
response = safe_requests.get(DUCKDUCKGO_URL.format(query_string=search_string), timeout=60)
if response.status_code == 200:
# print(response.text)
with open("result.json", "wb") as f:
Expand Down
Original file line number Diff line number Diff line change
@@ -1,10 +1,9 @@
import time

import requests
from security import safe_requests


def download(url):
return requests.get(url, timeout=60)
return safe_requests.get(url, timeout=60)


if __name__ == "__main__":
Expand Down
Original file line number Diff line number Diff line change
@@ -1,12 +1,11 @@
import time

import requests
from tomorrow import threads
from security import safe_requests


@threads(5)
def download(url):
return requests.get(url, timeout=5)
return safe_requests.get(url, timeout=5)


if __name__ == "__main__":
Expand Down
Loading
Loading