mirrors/github.com_openSUSE_osc
1
0
Fork 0
mirror of https://github.com/openSUSE/osc.git synced 2026-07-13 08:35:23 +02:00
Code Activity

Fix coding style in HttpSigner

Browse Source
This commit is contained in:
Daniel Mach
2026-03-06 10:58:48 +01:00
parent cbf98966b7
commit e465a6fdfa
4 changed files with 62 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/git-obs-quickstart.rst
+1 -1
View File
@@ -258,4 +258,4 @@ Known issues
- Reviews by groups/teams are not handled well.
If you approve, the team disappears and gets replaced with your login.
Then is not possible to search for such the team reviews and for example monitor
re-review requests during a team member's absence.
re-review requests during a team member's absence.
osc/commands_git/login_add.py
+5 -5
View File
@@ -41,12 +41,12 @@ class LoginAddCommand(osc.commandline_git.GitObsCommand):
self.parser.error("For SSH authentication, either --ssh-key or --ssh-agent must be specified together with --ssh-key-agent-pub")
elif (args.ssh_key and args.ssh_agent) or (args.ssh_key and args.token) or (args.ssh_agent and args.token):
self.parser.error("SSH authentication cannot be used together with token authentication, and --ssh-key and --ssh-agent cannot be used together")
ssh_login = (args.ssh_key or args.ssh_agent) and args.ssh_key_agent_pub
ssh_login = (args.ssh_key or args.ssh_agent) and args.ssh_key_agent_pub
if args.ssh_key:
from cryptography.hazmat.primitives import serialization
if not os.path.isfile(args.ssh_key):
self.parser.error(f"SSH key file '{args.ssh_key}' does not exist")
if not os.access(args.ssh_key, os.R_OK):
@@ -56,7 +56,7 @@ class LoginAddCommand(osc.commandline_git.GitObsCommand):
serialization.load_ssh_private_key(key_file.read(), password=None)
except Exception:
self.parser.error(f"SSH key file '{args.ssh_key}' is not a valid SSH private key")
if not ssh_login:
while not args.token or args.token == "-":
args.token = getpass.getpass(prompt=f"Enter Gitea token for user '{args.user}': ")
osc/commands_git/login_update.py
+8 -8
View File
@@ -49,13 +49,13 @@ class LoginUpdateCommand(osc.commandline_git.GitObsCommand):
final_ssh_key = args.new_ssh_key
# If setting a key, disable agent
final_ssh_agent = False
if new_ssh_agent is not None:
final_ssh_agent = new_ssh_agent
# If enabling agent, clear key
if new_ssh_agent:
final_ssh_key = None
if args.new_ssh_key_agent_pub:
final_ssh_key_agent_pub = args.new_ssh_key_agent_pub
@@ -73,13 +73,13 @@ class LoginUpdateCommand(osc.commandline_git.GitObsCommand):
if original_login_obj.ssh_agent and args.new_ssh_key:
print("Warning: switching from ssh-agent authentication to SSH key authentication, the ssh-agent setting will be disabled", file=sys.stderr)
if (final_ssh_key or final_ssh_agent) and args.new_token:
self.parser.error("Token authentication cannot be used together with SSH authentication")
if args.new_ssh_key:
from cryptography.hazmat.primitives import serialization
if not os.path.isfile(args.new_ssh_key):
self.parser.error(f"SSH key file '{args.new_ssh_key}' does not exist")
if not os.access(args.new_ssh_key, os.R_OK):
@@ -100,7 +100,7 @@ class LoginUpdateCommand(osc.commandline_git.GitObsCommand):
# TODO: try to authenticate to verify that the updated entry works
original_login_obj = self.gitea_conf.get_login(args.name)
final_ssh_key, final_ssh_agent, final_ssh_key_agent_pub = self._get_ssh_settings(args, original_login_obj)
if args.new_token == "-":
@@ -117,7 +117,7 @@ class LoginUpdateCommand(osc.commandline_git.GitObsCommand):
new_git_uses_http = True
else:
new_git_uses_http = None
if args.new_quiet in ("0", "no"):
new_quiet = False
elif args.new_quiet in ("1", "yes"):
@@ -138,7 +138,7 @@ class LoginUpdateCommand(osc.commandline_git.GitObsCommand):
new_quiet=new_quiet,
set_as_default=args.set_as_default,
)
print("")
print("Original entry:")
print(original_login_obj.to_human_readable_string())
osc/gitea_api/http_signer.py
+48 -56
View File
@@ -1,8 +1,8 @@
import datetime
import os
import base64
import datetime
import hashlib
import sys
import os
class HttpSigner:
def __init__(self, login_obj):
@@ -17,7 +17,7 @@ class HttpSigner:
from cryptography.hazmat.primitives import hashes
key_path = self.login_obj.ssh_key
if not os.path.exists(key_path):
raise FileNotFoundError(f"SSH private key not found at: {key_path}")
@@ -27,19 +27,16 @@ class HttpSigner:
algorithm_name = ""
if isinstance(private_key, ed25519.Ed25519PrivateKey):
algorithm_name = "ed25519"
def sign_func(data_bytes):
return private_key.sign(data_bytes)
elif isinstance(private_key, rsa.RSAPrivateKey):
algorithm_name = "rsa-sha512"
def sign_func(data_bytes):
return private_key.sign(
data_bytes,
padding.PKCS1v15(),
hashes.SHA512()
)
return private_key.sign(data_bytes, padding.PKCS1v15(), hashes.SHA512())
else:
raise ValueError(f"Unsupported key type: {type(private_key)}")
@@ -53,17 +50,17 @@ class HttpSigner:
try:
import paramiko
except ImportError:
raise ImportError("The 'paramiko' library is required for SSH Agent support. Please run: pip install paramiko")
raise ImportError("The 'paramiko' library is required for SSH Agent support.")
agent = paramiko.agent.Agent()
agent_keys = agent.get_keys()
if not agent_keys:
raise RuntimeError("No keys found in SSH Agent. Is it running and have you added keys (ssh-add)?")
target_fingerprint = self.login_obj.ssh_key_agent_pub
found_key = None
# Remove "SHA256:" prefix if present for calculation/comparison
target_fp_clean = target_fingerprint
if target_fp_clean.startswith("SHA256:"):
@@ -76,56 +73,55 @@ class HttpSigner:
fp_bytes = hashlib.sha256(key_blob).digest()
# OpenSSH uses standard base64 without padding usually, but python's b64encode adds padding.
# We need to strip standard padding '='
fp_str = base64.b64encode(fp_bytes).decode('ascii').rstrip('=')
fp_str = base64.b64encode(fp_bytes).decode("ascii").rstrip("=")
if fp_str == target_fp_clean:
found_key = key
break
if not found_key:
print("Available keys in agent:")
msg_lines = [f"Key with fingerprint {target_fingerprint} not found in SSH Agent."]
msg_lines += ["Available keys in the agent:"]
for key in agent_keys:
key_blob = key.asbytes()
fp_bytes = hashlib.sha256(key_blob).digest()
fp_str = base64.b64encode(fp_bytes).decode('ascii').rstrip('=')
print(f" - {key.get_name()} SHA256:{fp_str}")
raise ValueError(f"Key with fingerprint {target_fingerprint} not found in SSH Agent.")
#print(f"Using SSH Agent key: {found_key.get_name()}")
fp_str = base64.b64encode(fp_bytes).decode("ascii").rstrip("=")
msg_lines += [f" - {key.get_name()} SHA256:{fp_str}"]
raise ValueError("\n".join(msg_lines))
# Determine algorithm name
# Paramiko's get_name() usually returns 'ssh-rsa', 'ssh-ed25519', etc.
algo_map = {
"ssh-ed25519": "ed25519",
"ssh-rsa": "rsa-sha512", # We assume we want sha512 for RSA
"ssh-rsa": "rsa-sha512", # We assume we want sha512 for RSA
"rsa-sha2-512": "rsa-sha512",
"rsa-sha2-256": "rsa-sha256"
"rsa-sha2-256": "rsa-sha256",
}
# Default to the key type, mapped or raw
algorithm_name = algo_map.get(found_key.get_name(), found_key.get_name())
def sign_func(data_bytes):
# agent.sign_data returns a Message object or bytes containing the SSH signature blob
# The SSH signature blob is: [4 byte len][algo name][4 byte len][signature]
# We need to send the signature part.
# Note: For RSA, we might need to specify flags to get SHA512.
# Paramiko's sign_ssh_data doesn't easily expose flags in the high level AgentKey API
# in older versions, but let's try just signing.
# in older versions, but let's try just signing.
# Modern agents usually negotiate or we might just get what we get.
# If it is RSA, we really want rsa-sha2-512.
# But paramiko AgentKey.sign_ssh_data just calls the agent.
# result is a paramiko.message.Message or bytes
# It typically contains the entire blob: [string algo][string sig]
# Use key.sign_ssh_data(data)
sig_result = found_key.sign_ssh_data(data_bytes)
# sig_result is usually a paramiko.Message object.
if hasattr(sig_result, 'rewind'):
if hasattr(sig_result, "rewind"):
sig_result.rewind()
# The structure is: string(algo), string(signature_blob)
sig_algo = sig_result.get_string()
@@ -136,20 +132,21 @@ class HttpSigner:
# SSH string format: 4 bytes len, string data
# We skip the algo name
import struct
if not isinstance(sig_result, bytes):
raise TypeError(f"Unexpected return type from sign_ssh_data: {type(sig_result)}")
offset = 0
algo_len = struct.unpack('>I', sig_result[offset:offset+4])[0]
algo_len = struct.unpack(">I", sig_result[offset : offset + 4])[0]
offset += 4 + algo_len
sig_len = struct.unpack('>I', sig_result[offset:offset+4])[0]
sig_len = struct.unpack(">I", sig_result[offset : offset + 4])[0]
offset += 4
sig_blob = sig_result[offset:offset+sig_len]
sig_blob = sig_result[offset : offset + sig_len]
return sig_blob
return sign_func, algorithm_name
def get_signed_header(self, method, path):
"""
Sign the request data using the configured authentication method (SSH key or agent).
@@ -161,7 +158,7 @@ class HttpSigner:
sign_func, algorithm_name = self._get_signer_from_agent()
else:
raise ValueError("No SSH authentication method configured for this login entry.")
# Timestamps
now = datetime.datetime.now(datetime.timezone.utc)
created = int(now.timestamp())
@@ -177,34 +174,29 @@ class HttpSigner:
path += f"?{parsed.query}"
request_target_value = f"{method.lower()} {path}"
#print(f"Signing request target: {request_target_value}")
signing_string = (
f"(request-target): {request_target_value}\n"
f"(created): {created}\n"
f"(expires): {expires}"
)
try:
signature_bytes = sign_func(signing_string.encode('utf-8'))
signature_b64 = base64.b64encode(signature_bytes).decode('utf-8')
signature_bytes = sign_func(signing_string.encode("utf-8"))
signature_b64 = base64.b64encode(signature_bytes).decode("utf-8")
except Exception as e:
raise RuntimeError(f"Failed to sign data: {e}")
headers_list = "(request-target) (created) (expires)"
signature_header = (
f'keyId="{self.login_obj.ssh_key_agent_pub}",'
f'algorithm="{algorithm_name}",'
f'headers="{headers_list}",'
f'signature="{signature_b64}",'
f'created={created},'
f'expires={expires}'
f"created={created},"
f"expires={expires}"
)
headers = {
'Signature': signature_header
}
return headers
headers = {"Signature": signature_header}
return headers