mirror of
https://gitea.publichub.eu/oscar.krause/fastapi-dls.git
synced 2025-11-03 20:06:07 +00:00
migrated from "pycryptodome" to "cryptography"
This commit is contained in:
16
app/main.py
16
app/main.py
@@ -21,7 +21,7 @@ from starlette.middleware.cors import CORSMiddleware
|
||||
from starlette.responses import StreamingResponse, JSONResponse as JSONr, HTMLResponse as HTMLr, Response, RedirectResponse
|
||||
|
||||
from orm import Origin, Lease, init as db_init, migrate
|
||||
from util import load_key, load_file
|
||||
from util import load_private_key, load_public_key, get_pem, load_file
|
||||
|
||||
# Load variables
|
||||
load_dotenv('../version.env')
|
||||
@@ -42,8 +42,8 @@ DLS_PORT = int(env('DLS_PORT', '443'))
|
||||
SITE_KEY_XID = str(env('SITE_KEY_XID', '00000000-0000-0000-0000-000000000000'))
|
||||
INSTANCE_REF = str(env('INSTANCE_REF', '10000000-0000-0000-0000-000000000001'))
|
||||
ALLOTMENT_REF = str(env('ALLOTMENT_REF', '20000000-0000-0000-0000-000000000001'))
|
||||
INSTANCE_KEY_RSA = load_key(str(env('INSTANCE_KEY_RSA', join(dirname(__file__), 'cert/instance.private.pem'))))
|
||||
INSTANCE_KEY_PUB = load_key(str(env('INSTANCE_KEY_PUB', join(dirname(__file__), 'cert/instance.public.pem'))))
|
||||
INSTANCE_KEY_RSA = load_private_key(str(env('INSTANCE_KEY_RSA', join(dirname(__file__), 'cert/instance.private.pem'))))
|
||||
INSTANCE_KEY_PUB = load_public_key(str(env('INSTANCE_KEY_PUB', join(dirname(__file__), 'cert/instance.public.pem'))))
|
||||
TOKEN_EXPIRE_DELTA = relativedelta(days=int(env('TOKEN_EXPIRE_DAYS', 1)), hours=int(env('TOKEN_EXPIRE_HOURS', 0)))
|
||||
LEASE_EXPIRE_DELTA = relativedelta(days=int(env('LEASE_EXPIRE_DAYS', 90)), hours=int(env('LEASE_EXPIRE_HOURS', 0)))
|
||||
LEASE_RENEWAL_PERIOD = float(env('LEASE_RENEWAL_PERIOD', 0.15))
|
||||
@@ -51,8 +51,8 @@ LEASE_RENEWAL_DELTA = timedelta(days=int(env('LEASE_EXPIRE_DAYS', 90)), hours=in
|
||||
CLIENT_TOKEN_EXPIRE_DELTA = relativedelta(years=12)
|
||||
CORS_ORIGINS = str(env('CORS_ORIGINS', '')).split(',') if (env('CORS_ORIGINS')) else [f'https://{DLS_URL}']
|
||||
|
||||
jwt_encode_key = jwk.construct(INSTANCE_KEY_RSA.export_key().decode('utf-8'), algorithm=ALGORITHMS.RS256)
|
||||
jwt_decode_key = jwk.construct(INSTANCE_KEY_PUB.export_key().decode('utf-8'), algorithm=ALGORITHMS.RS256)
|
||||
jwt_encode_key = jwk.construct(get_pem(INSTANCE_KEY_RSA), algorithm=ALGORITHMS.RS256)
|
||||
jwt_decode_key = jwk.construct(get_pem(INSTANCE_KEY_PUB), algorithm=ALGORITHMS.RS256)
|
||||
|
||||
# Logging
|
||||
LOG_LEVEL = logging.DEBUG if DEBUG else logging.INFO
|
||||
@@ -264,10 +264,10 @@ async def _client_token():
|
||||
},
|
||||
"service_instance_public_key_configuration": {
|
||||
"service_instance_public_key_me": {
|
||||
"mod": hex(INSTANCE_KEY_PUB.public_key().n)[2:],
|
||||
"exp": int(INSTANCE_KEY_PUB.public_key().e),
|
||||
"mod": hex(INSTANCE_KEY_PUB.public_numbers().n)[2:],
|
||||
"exp": int(INSTANCE_KEY_PUB.public_numbers().e),
|
||||
},
|
||||
"service_instance_public_key_pem": INSTANCE_KEY_PUB.export_key().decode('utf-8'),
|
||||
"service_instance_public_key_pem": get_pem(INSTANCE_KEY_PUB).decode('utf-8'),
|
||||
"key_retention_mode": "LATEST_ONLY"
|
||||
},
|
||||
}
|
||||
|
||||
56
app/util.py
56
app/util.py
@@ -11,31 +11,51 @@ def load_file(filename: str) -> bytes:
|
||||
return content
|
||||
|
||||
|
||||
def load_key(filename: str) -> "RsaKey":
|
||||
try:
|
||||
# Crypto | Cryptodome on Debian
|
||||
from Crypto.PublicKey import RSA
|
||||
from Crypto.PublicKey.RSA import RsaKey
|
||||
except ModuleNotFoundError:
|
||||
from Cryptodome.PublicKey import RSA
|
||||
from Cryptodome.PublicKey.RSA import RsaKey
|
||||
def load_private_key(filename: str) -> "RSAPrivateKey":
|
||||
from cryptography.hazmat.primitives.serialization import load_pem_private_key
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
log.debug(f'Importing RSA-Key from "{filename}"')
|
||||
return RSA.import_key(extern_key=load_file(filename), passphrase=None)
|
||||
|
||||
with open(filename, 'rb') as f:
|
||||
data = f.read()
|
||||
return load_pem_private_key(data.strip(), password=None)
|
||||
|
||||
|
||||
def generate_key() -> "RsaKey":
|
||||
try:
|
||||
# Crypto | Cryptodome on Debian
|
||||
from Crypto.PublicKey import RSA
|
||||
from Crypto.PublicKey.RSA import RsaKey
|
||||
except ModuleNotFoundError:
|
||||
from Cryptodome.PublicKey import RSA
|
||||
from Cryptodome.PublicKey.RSA import RsaKey
|
||||
def load_public_key(filename: str) -> "RSAPublicKey":
|
||||
from cryptography.hazmat.primitives.serialization import load_pem_public_key
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
log.debug(f'Importing RSA-Key from "{filename}"')
|
||||
|
||||
with open(filename, 'rb') as f:
|
||||
data = f.read()
|
||||
return load_pem_public_key(data.strip())
|
||||
|
||||
|
||||
def get_pem(key) -> bytes | None:
|
||||
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey, RSAPublicKey
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
|
||||
if isinstance(key, RSAPrivateKey):
|
||||
return key.private_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.TraditionalOpenSSL,
|
||||
encryption_algorithm=serialization.NoEncryption()
|
||||
)
|
||||
if isinstance(key, RSAPublicKey):
|
||||
return key.public_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PublicFormat.SubjectPublicKeyInfo
|
||||
)
|
||||
|
||||
|
||||
def generate_private_key() -> "RSAPrivateKey":
|
||||
from cryptography.hazmat.primitives.asymmetric import rsa
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
log.debug(f'Generating RSA-Key')
|
||||
return RSA.generate(bits=2048)
|
||||
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||||
|
||||
|
||||
class NV:
|
||||
|
||||
Reference in New Issue
Block a user