import base64
import hashlib
import hmac
import datetime
import requests
import xmltodict
import os
import re
account_name = 'name'
account_key = "key"
share_name = 'test'
directory_path = 'dir'
api_version = "2021-04-10"
request_time = datetime.datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')
def convert_query_params(query_string):
# クエリパラメータを `&` で分割
params = query_string.split('&')
# 各パラメータを `=` で分割し、辞書に変換
param_dict = {}
for param in params:
if '=' in param:
key, value = param.split('=', 1) # `=` で分割、1回の分割で値を取得
param_dict[key] = value
# 辞書をキーでソートし、`key:value` 形式に変換
sorted_params = [f"{key}:{param_dict[key]}" for key in sorted(param_dict.keys())]
# `\n` で連結
result = '\n'.join(sorted_params)
return result
def get_file_name_from_url(url):
# ファイル名のパターンを定義（例: .txt, .jpg, .pngなどの拡張子）
file_pattern = re.compile(r'/([^/]+\.[a-zA-Z0-9]+)$')
# URLのパス部分を抽出
path = url.split('?')[0]
# パターンに一致するかどうかを判定
match = file_pattern.search(path)
# 一致する場合はファイル名を返し、一致しない場合は空文字を返す
return match.group(1) if match else ''
def generate_signature(method, url, account_key):
print(str(os.path.getsize("src/azure/create.txt")))
signature_string_values = {
'VERB': method,
'Content-Encoding': '',
'Content-Language': '',
'Content-Length': str(os.path.getsize("src/azure/create.txt")),
'Content-MD5': '',
'Content-Type': '',
'Date': '',
'If-Modified-Since': '',
'If-Match': '',
'If-None-Match': '',
'If-Unmodified-Since': '',
'Range': '',
'CanonicalizedHeaders': '',
'CanonicalizedResource': ''
}
# URLのパス部分とクエリパラメータを分ける
url_path = url.split('?')[0]
query_params = url.split('?')[1] if '?' in url else ''
# When we just want to list details for the File share, we only need:
signature_string_values['VERB'] = method
signature_string_values['CanonicalizedHeaders'] = (
'x-ms-content-length:1024\n'
'x-ms-content-type:text/plain\n'
'x-ms-date:' + request_time + '\n'
'x-ms-file-attributes:Archive\n'
'x-ms-file-creation-time:now\n'
'x-ms-file-last-write-time:now\n'
'x-ms-file-permission:inherit\n'
'x-ms-type:file\n'
'x-ms-version:' + api_version + "\n"
)
file_name = get_file_name_from_url(url)
if file_name:
signature_string_values['CanonicalizedResource'] = f"/{account_name}/{share_name}/{directory_path}/{file_name}"
else:
print("none file")
signature_string_values['CanonicalizedResource'] = f"/{account_name}/{share_name}/{directory_path}/\n{convert_query_params(query_params)}"
string_to_sign = (signature_string_values['VERB'] + '\n'
+ signature_string_values['Content-Encoding'] + '\n'
+ signature_string_values['Content-Language'] + '\n'
+ signature_string_values['Content-Length'] + '\n'
+ signature_string_values['Content-MD5'] + '\n'
+ signature_string_values['Content-Type'] + '\n'
+ signature_string_values['Date'] + '\n'
+ signature_string_values['If-Modified-Since'] + '\n'
+ signature_string_values['If-Match'] + '\n'
+ signature_string_values['If-None-Match'] + '\n'
+ signature_string_values['If-Unmodified-Since'] + '\n'
+ signature_string_values['Range'] + '\n'
+ signature_string_values['CanonicalizedHeaders']
+ signature_string_values['CanonicalizedResource'])
# 署名を生成する
# We need an to decode our storage account key
key = base64.b64decode(account_key.encode('utf-8'))
# Now let's sign the string
signature = base64.b64encode(hmac.new(key, msg=string_to_sign.encode('utf-8'), digestmod = hashlib.sha256).digest()).decode()
return signature
# リクエストの構成
url = f"https://{account_name}.file.core.windows.net/{share_name}/{directory_path}/test.txt"
signature = generate_signature('PUT', url,account_key)
headers = {
'Authorization': f"SharedKey {account_name}:{signature}",
'x-ms-content-length': '1024',
'x-ms-content-type': 'text/plain',
'x-ms-date': request_time,
'x-ms-file-attributes': 'Archive',
'x-ms-file-creation-time': 'now',
'x-ms-file-last-write-time': 'now',
'x-ms-file-permission': 'inherit',
'x-ms-type': 'file',
'x-ms-version': api_version,
}
# リクエストを送信
with open("src/azure/create.txt",'rb') as file:
print(file)
response = requests.put(url, headers=headers,data=file)
print(f"URL: {url}")
print(response.status_code)
print(response.text)