Compare commits
1 Commits
feat_signu
...
dependabot
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
971bdd51d2 |
@@ -14,8 +14,7 @@
|
||||
|
||||
// Features to add to the dev container. More info: https://containers.dev/features.
|
||||
"features": {
|
||||
"ghcr.io/devcontainers/features/docker-in-docker:2": {},
|
||||
"ghcr.io/warrenbuckley/codespace-features/sqlite:1": {}
|
||||
"ghcr.io/devcontainers/features/docker-in-docker:2": {}
|
||||
},
|
||||
|
||||
// Use 'forwardPorts' to make a list of ports inside the container available locally.
|
||||
|
||||
@@ -43,7 +43,7 @@ shellingham==1.5.4
|
||||
sniffio==1.3.1
|
||||
SQLAlchemy==2.0.44
|
||||
sqlmodel==0.0.27
|
||||
starlette==0.48.0
|
||||
starlette==0.49.1
|
||||
typer==0.19.2
|
||||
typing-inspection==0.4.2
|
||||
typing_extensions==4.15.0
|
||||
|
||||
1
slopserver/common.py
Normal file
1
slopserver/common.py
Normal file
@@ -0,0 +1 @@
|
||||
TEMP_HMAC_KEY = "0460de065912d0292df1e7422a5ed2dc362ed56d6bab64fe50b89957463061f3"
|
||||
@@ -8,7 +8,7 @@ from altcha import Payload as AltchaPayload, verify_solution
|
||||
|
||||
from urllib.parse import urlparse, ParseResult
|
||||
|
||||
from slopserver.settings import settings
|
||||
from slopserver.common import TEMP_HMAC_KEY
|
||||
|
||||
NAMING_CONVENTION = {
|
||||
"ix": "ix_%(column_0_label)s",
|
||||
@@ -73,7 +73,8 @@ def url_validator(urls: list[str]) -> list[ParseResult]:
|
||||
return parsed_urls
|
||||
|
||||
def altcha_validator(altcha_response: AltchaPayload):
|
||||
verified = verify_solution(altcha_response, settings.altcha_secret)
|
||||
# verified = verify_solution(altcha_response, TEMP_HMAC_KEY)
|
||||
verified = (True, None)
|
||||
if not verified[0]:
|
||||
raise ValueError(f"altcha verification failed: {verified[1]}")
|
||||
return None
|
||||
@@ -85,4 +86,4 @@ class SlopReport(BaseModel):
|
||||
class SignupForm(BaseModel):
|
||||
email: EmailStr
|
||||
password: SecretStr
|
||||
altcha: Annotated[str, AfterValidator(altcha_validator)]
|
||||
altcha: Annotated[Base64Str, AfterValidator(altcha_validator)]
|
||||
@@ -17,7 +17,7 @@ from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from pydantic import AfterValidator, Base64Str
|
||||
|
||||
from pydantic_settings import BaseSettings
|
||||
|
||||
from sqlalchemy import create_engine
|
||||
|
||||
@@ -29,15 +29,23 @@ import jwt
|
||||
|
||||
from uuid import uuid4
|
||||
|
||||
from slopserver.settings import settings
|
||||
|
||||
from slopserver.models import Domain, Path, User
|
||||
from slopserver.models import SlopReport, SignupForm, altcha_validator
|
||||
from slopserver.db import select_slop, insert_slop, get_user, create_user
|
||||
from slopserver.common import TEMP_HMAC_KEY
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
||||
|
||||
class ServerSettings(BaseSettings):
|
||||
db_url: str = "sqlite+pysqlite:///test_db.sqlite"
|
||||
token_secret: str = "5bcc778a96b090c3ac1d587bb694a060eaf7bdb5832365f91d5078faf1fff210"
|
||||
# altcha_secret: str
|
||||
|
||||
settings = ServerSettings()
|
||||
|
||||
|
||||
DB_ENGINE = create_engine(settings.db_url)
|
||||
TOKEN_SECRET = settings.token_secret
|
||||
@@ -126,14 +134,14 @@ async def signup_form(form_data: Annotated[SignupForm, Form()]):
|
||||
raise HTTPException(status_code=409, detail="User already exists")
|
||||
|
||||
# create user
|
||||
create_user(form_data.email, get_password_hash(form_data.password.get_secret_value()), DB_ENGINE)
|
||||
create_user(form_data.email, get_password_hash(form_data.password), DB_ENGINE)
|
||||
|
||||
@app.get("/altcha-challenge")
|
||||
async def altcha_challenge():
|
||||
options = ChallengeOptions(
|
||||
expires=datetime.now() + timedelta(minutes=10),
|
||||
max_number=80000,
|
||||
hmac_key=settings.altcha_secret
|
||||
max_number=100000,
|
||||
hmac_key=TEMP_HMAC_KEY
|
||||
)
|
||||
challenge = create_challenge(options)
|
||||
return challenge
|
||||
@@ -146,5 +154,26 @@ async def simple_login(username: Annotated[str, Form()], password: Annotated[str
|
||||
token = generate_auth_token(username)
|
||||
return {"access_token": token, "token_type": "bearer"}
|
||||
|
||||
@app.post("/altcha-challenge")
|
||||
async def altcha_verify(payload: Annotated[Base64Str, AfterValidator(altcha_validator)]):
|
||||
# if verified, return a JWT for anonymous API access
|
||||
expiration = datetime.now() + timedelta(days=30)
|
||||
uuid = uuid4()
|
||||
bearer_token = {
|
||||
"iss": "slopserver",
|
||||
"exp": int(expiration.timestamp()),
|
||||
"aud": "slopserver",
|
||||
"sub": str(uuid),
|
||||
"client_id": str(uuid),
|
||||
"iat": int(datetime.now().timestamp()),
|
||||
"jti": str(uuid)
|
||||
}
|
||||
|
||||
encoded_jwt = jwt.encode(bearer_token, TOKEN_SECRET, ALGO)
|
||||
|
||||
return encoded_jwt
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
@@ -1,9 +0,0 @@
|
||||
from pydantic_settings import BaseSettings
|
||||
|
||||
|
||||
class ServerSettings(BaseSettings):
|
||||
db_url: str = "sqlite+pysqlite:///test_db.sqlite"
|
||||
token_secret: str = "5bcc778a96b090c3ac1d587bb694a060eaf7bdb5832365f91d5078faf1fff210"
|
||||
altcha_secret: str = "0460de065912d0292df1e7422a5ed2dc362ed56d6bab64fe50b89957463061f3"
|
||||
|
||||
settings = ServerSettings()
|
||||
Reference in New Issue
Block a user