11#
22# SPDX-FileCopyrightText: 2023 Nextcloud GmbH and Nextcloud contributors
33# SPDX-License-Identifier: AGPL-3.0-or-later
4- #
5- from typing import Annotated
64
7- from fastapi import Depends , FastAPI
8- from fastapi .responses import JSONResponse
5+ import json
6+ import os
7+ from base64 import b64decode , b64encode
8+ from urllib .request import Request , urlopen
99
10- from nc_py_api import NextcloudApp , ex_app
10+ from fastapi import FastAPI , HTTPException , Request as FastAPIRequest
11+ from fastapi .responses import JSONResponse
1112
13+ APP_ID = os .environ ["APP_ID" ]
14+ APP_SECRET = os .environ ["APP_SECRET" ]
15+ APP_VERSION = os .environ .get ("APP_VERSION" , "1.0.0" )
16+ NEXTCLOUD_URL = os .environ .get ("NEXTCLOUD_URL" , "http://localhost:8080" ).rstrip ("/" )
1217
1318APP = FastAPI ()
1419
1520
21+ def verify_auth (request : FastAPIRequest ) -> str :
22+ """Validate AppAPI auth headers. Returns the username on success."""
23+ ex_app_id = request .headers .get ("EX-APP-ID" , "" )
24+ ex_app_version = request .headers .get ("EX-APP-VERSION" , "" )
25+ auth_app_api = request .headers .get ("AUTHORIZATION-APP-API" , "" )
26+
27+ missing = [h for h , v in [
28+ ("EX-APP-ID" , ex_app_id ),
29+ ("EX-APP-VERSION" , ex_app_version ),
30+ ("AUTHORIZATION-APP-API" , auth_app_api ),
31+ ] if not v ]
32+ if missing :
33+ raise HTTPException (status_code = 401 , detail = f"Missing headers: { missing } " )
34+
35+ if ex_app_id != APP_ID :
36+ raise HTTPException (status_code = 401 , detail = f"Invalid EX-APP-ID: { ex_app_id } " )
37+
38+ try :
39+ decoded = b64decode (auth_app_api ).decode ("UTF-8" )
40+ username , secret = decoded .split (":" , maxsplit = 1 )
41+ except Exception :
42+ raise HTTPException (status_code = 401 , detail = "Malformed AUTHORIZATION-APP-API" )
43+
44+ if secret != APP_SECRET :
45+ raise HTTPException (status_code = 401 , detail = "Invalid app secret" )
46+
47+ return username
48+
49+
50+ def log_to_nextcloud (username : str , level : int , message : str ) -> None :
51+ """Send a log entry to Nextcloud via OCS API."""
52+ url = f"{ NEXTCLOUD_URL } /ocs/v1.php/apps/app_api/api/v1/log"
53+ data = json .dumps ({"level" : level , "message" : message }).encode ("UTF-8" )
54+ auth_header = b64encode (f"{ username } :{ APP_SECRET } " .encode ("UTF-8" )).decode ("ASCII" )
55+ req = Request (url , data = data , method = "POST" , headers = {
56+ "Content-Type" : "application/json" ,
57+ "OCS-APIRequest" : "true" ,
58+ "EX-APP-ID" : APP_ID ,
59+ "EX-APP-VERSION" : APP_VERSION ,
60+ "AA-VERSION" : "2.0.0" ,
61+ "AUTHORIZATION-APP-API" : auth_header ,
62+ })
63+ try :
64+ urlopen (req )
65+ except Exception as e :
66+ print (f"[test-app] Failed to log to Nextcloud: { e } " )
67+
68+
1669@APP .put ("/enabled" )
17- async def enabled_callback (
18- enabled : bool ,
19- nc : Annotated [NextcloudApp , Depends (ex_app .nc_app )],
20- ):
70+ async def enabled_callback (enabled : bool , request : FastAPIRequest ):
71+ username = verify_auth (request )
2172 if enabled :
22- nc . log ( ex_app . LogLvl . WARNING , f"Hello from { nc . app_cfg . app_name } :)" )
73+ log_to_nextcloud ( username , 2 , f"Hello from { APP_ID } :)" )
2374 else :
24- nc . log ( ex_app . LogLvl . WARNING , f"Bye bye from { nc . app_cfg . app_name } :(" )
75+ log_to_nextcloud ( username , 2 , f"Bye bye from { APP_ID } :(" )
2576 return JSONResponse (content = {"error" : "" }, status_code = 200 )
2677
2778
@@ -31,4 +82,10 @@ async def heartbeat_callback():
3182
3283
3384if __name__ == "__main__" :
34- ex_app .run_app ("install_no_init:APP" , log_level = "trace" )
85+ import uvicorn
86+ uvicorn .run (
87+ "install_no_init:APP" ,
88+ host = os .environ .get ("APP_HOST" , "127.0.0.1" ),
89+ port = int (os .environ ["APP_PORT" ]),
90+ log_level = "trace" ,
91+ )
0 commit comments