RAG에서 Reqs로: 그래프와 챗봇으로 ASVS에 쉽게 접근하기

OWASP Application Security Verification Standard (ASVS)는 훌륭하지만, OWASP Top 10에 비해 덜 사용되는 것 같아요. 개념 자체는 정말 환상적이죠. 모든 개발 조직이 애플리케이션 보안을 자체적으로 평가하고 체계적으로 개선하는 데 사용할 수 있는 표준이니까요. 그런데 왜 이렇게 좋은 표준이 널리 사용되지 않을까요?

이 문제에 대해 Josh Grossman의 이야기를 한번 들어볼까요? 그는 ASVS 프로젝트의 열렬한 지지자 중 한 명인데요, 이런 말을 했어요. "ASVS에는 약 280개의 요구 사항이 있어요. 모든 기능 하나하나에 대한 요구 사항을 일일이 살펴보고 싶진 않죠. 중요한 것에 집중하고 싶고, 애플리케이션 개발의 특정 단계에서 어떤 기능이 중요한지 집중하고 싶을 뿐이에요."

이번 블로그에서는 Neo4j 엔지니어링 팀이 ASVS를 도입하면서 겪었던 여정을 이야기해 보려고 해요.

배경

ASVS를 시작할 때 가장 먼저 권장되는 사항 중 하나는 조직의 상황에 맞게 조정하는 거예요. 그래서 표준을 검토하고 관련 없는 요구 사항을 제거하는 작업을 자원한 보안 챔피언이 있었죠. 이 과정을 거친 후에도 여전히 166개의 유효한 요구 사항이 남아있었어요. 보안 팀에서 개발자에게 모든 기능에 대해 이 긴 목록을 검토하도록 요청할 수 있을까요? 물론 요청은 할 수 있겠지만, 현실적으로 열광적인 "네!"라는 대답을 기대하긴 어렵겠죠. 요구 사항이 너무 많으니까요. 게다가, 실제 적용 측면에서 어떻게 해야 할까요? 스프레드시트 말고 다른 방법이 필요하잖아요. 분명히 더 나은 방법이 있을 거예요.

영감

저희는 Neo4j의 작은 AppSec 팀이에요. 하지만 ASVS가 본질적으로 매우 "그래프" 데이터라는 것을 깨닫기 위해 꼭 저희 회사에 다닐 필요는 없을 거예요. 이 점을 저희에게 유리하게 활용할 수 있을까요?

ASVS를 Neo4j에 그래프로 가져옴

ASVS 도입 방법을 고민하기 시작했을 때, 업계에서는 LLM(Large Language Model)이 엄청난 성장세를 보이고 있었어요. LLM을 사용해서 기능에 대한 설명을 입력하고, 관련된 ASVS 요구 사항을 생성하도록 할 수 있지 않을까요?

첫 번째 시도 – 환각

요구 사항이 설명된 기능과 얼마나 관련 있는지 논의하기도 전에 ChatGPT가 요구 사항에 대해 환각을 일으키고 있었고, Prompt Engineering으로도 현실을 고수하도록 설득할 수 없다는 게 분명해졌어요.

다행히 여러 블로그에서 Retrieval-Augmented Generation(RAG)을 사용하여 LLM 기반을 논의했죠. RAG는 LLM의 출력을 최적화하는 프로세스인데, 응답을 생성하기 전에 훈련 데이터 소스 외부의 권위 있는 Knowledge Graph를 참조하는 방식이에요.

저희는 ASVS에 대한 권위 있는 Knowledge Graph를 가지고 있고, Vector Index 및 유사성 검색을 지원하는 데이터베이스도 가지고 있죠. 그리고 강의 시간에 RAG 사용법을 설명하기 위해 영화 데이터베이스를 사용하기도 했어요. 전부 합쳐보면 어떨까요?

도구 아키텍처

저희는 사용자가 Jira 또는 Trello 카드에 입력하는 것처럼 한두 문단으로 기능을 설명하길 원해요. 도구는 이 텍스트의 Vector Embedding을 계산하고, 유사성 임계값을 초과하는 ASVS 요구 사항을 검색하고, 이러한 요구 사항을 LLM에 컨텍스트로 제공할 거예요.

코드와 상호 작용의 예를 보여주기 전에 기본적인 Knowledge Graph를 자세히 살펴볼게요. (이 레이아웃에 대해 궁금하시다면, 작은 Node는 레벨 1 요구 사항을 나타낸답니다.)

V7 확대: 오류 처리 및 로깅

주요 관계는 다음과 같아요.(c:Control)-[:INCLUDES]->(a:Area)-[:HAS]->(r:Requirement)

Control and Area Node는 매우 간단해요. 번호와 이름만 있답니다.

Requirement 노드가 좀 더 보기 좋네요!

Requirement 노드 속성

이제 Vector Embedding을 계산할 수 있어요. Verification Requirement 속성을 저장하고 이를 별도의 노드에 저장하는 거죠.

from neo4j import GraphDatabase, Query
from openai.embeddings_utils import get_embedding
import openai
import os


"""
LoadEmbedding: call OpenAI embedding API to generate embeddings for each property of node in Neo4j
Version: 1.1
"""


EMBEDDING_MODEL = "text-embedding-ada-002"


NEO4J_URL = os.environ['NEO4J_URI']
NEO4J_USER = os.environ['NEO4J_USER']
NEO4J_PASSWORD = os.environ['NEO4J_PASS']
OPENAI_KEY = os.environ['OPENAI_KEY']




class LoadEmbedding:
   def __init__(self, uri, user, password):
       self.driver = GraphDatabase.driver(uri, auth=(user, password))
       openai.api_key = OPENAI_KEY


   def close(self):
       self.driver.close()


   def load_embedding_to_node_property(self, node_label, node_property):
       self.driver.verify_connectivity()


       with self.driver.session(database="neo4j") as session:


           result = session.run(f"""
               MATCH (a:{node_label})
               WHERE a.{node_property} IS NOT NULL
               RETURN id(a) AS id, a.{node_property} AS node_property
           """)




           # call OpenAI embedding API to generate embeddings for each property of node
           # for each node, update the embedding property


           for record in result.data():
               id = record["id"]
               text = record["node_property"]
               # Below, instead of using the text as the input for embedding, we add label and property name in
               # front of it
               embedding = get_embedding(f"{node_label} {node_property} - {text}", EMBEDDING_MODEL)
               # key property of Embedding node differentiates different embeddings
               cypher = "CREATE (e:Embedding) SET e.key=$key, e.value=$embedding"
               cypher = cypher + " WITH e MATCH (n) WHERE id(n) = $id CREATE (n) -[:HAS_EMBEDDING]-> (e)"
               session.run(cypher, key=node_property, embedding=embedding, id=id)




if __name__ == "__main__":
   loader = LoadEmbedding(NEO4J_URL, NEO4J_USER, NEO4J_PASSWORD)
   loader.load_embedding_to_node_property("Requirement", "`Verification Requirement`")


   loader.close()

이 스크립트를 실행한 후에는 반드시 index를 생성해야 해요.

CALL db.index.vector.createNodeIndex('embeddingIndex', 'Embedding', 'value', 1536, 'COSINE')

사전 계산된 embedding과 index가 포함된 Aura 데이터베이스 덤프는 여기서 다운로드할 수 있어요: .

이제 Knowledge Graph가 준비되었으니, 이걸 모두 통합해 볼까요? 인터페이스로는 를 사용해볼게요.

import os
import streamlit as st
from openai import OpenAI
from langchain_community.graphs import Neo4jGraph


NEO4J_URL = os.environ['NEO4J_URI']
NEO4J_USER = os.environ['NEO4J_USER']
NEO4J_PASSWORD = os.environ['NEO4J_PASS']
OPENAI_KEY = os.environ['OPENAI_API_KEY']


EMBEDDING_MODEL = "text-embedding-ada-002"
OPENAI_MODEL = "gpt-4o"  # replace with your favourite model


st.title('🦜🔗 AppSec Verification Standard (ASVS) Chat App')
st.write("This program looks up ASVS requirements relevant to your feature description.")
with st.sidebar:
   st.subheader("Browse the full ASVS:")
   st.write("https://github.com/OWASP/ASVS/tree/master/4.0/en")
   st.subheader("Here is an example of feature description:")
   st.markdown('''_We want to add some pay-per-use features in our SaaS product. Each cloud tenant will have
    an agent monitoring the usage of these features. The agents send data through a custom proxy to BigQuery.
    We will use the data in BigQuеry to bill the customers._
   ''')
   st.subheader("Some params you can tweak:")
   num_reqs = st.number_input("Maximum number of ASVS requirements to take into account", 2, 20, value=10,
                               help='''Number of reqs above similarity threshold to pass to OpenAI."
                                 If you have lots of relevant requirements, increase this param to pass them on.''')
   threshold = st.number_input("Cut-off for low scoring matches", 0.5, 0.99, value=0.87,
                               help='''Semantic similarity threshold in vector lookup, 0.99 is the most strict.
                                       0.85 seems reasonable, but YMMV. Passing in irrelevant reqs will lead to
                                       hallucinations.''')


   st.subheader("Bring your own key:")
   open_api_key = st.text_input("OpenAI API Key", type="password")  # Not overwriting env. key at the moment


llm = OpenAI()


graph = Neo4jGraph(
   url=NEO4J_URL,
   username=NEO4J_USER,
   password=NEO4J_PASSWORD
)


# Set a default model
if "openai_model" not in st.session_state:
   st.session_state["openai_model"] = OPENAI_MODEL


# Initialize chat history
if "messages" not in st.session_state:
   st.session_state.messages = []


# Display chat messages from history on app rerun
for message in st.session_state.messages:
   with st.chat_message(message["role"]):
       st.markdown(message["content"])




def generate_response(input_text):
   embedding = llm.embeddings.create(input=input_text, model=EMBEDDING_MODEL)


   r = graph.query("""


       // 2. Find other requirements which have high semantic similarity on description
       CALL db.index.vector.queryNodes("embeddingIndex", $lookup_num, $feature_embedding) YIELD node, score
       WITH node, score
       WHERE score > $threshold // exclude low-scoring matches


       // 3. From the returned Embedding nodes, find their connected Requirement nodes
       MATCH (m2:Requirement)-[:HAS_EMBEDDING]->(node)
       WHERE node.key = '`Verification Requirement`'


       RETURN  m2.`#` AS reqNumber, m2.`Verification Requirement` AS requirement, score
       ORDER BY score DESC;""",
                   {'feature_embedding': embedding.data[0].embedding,
                           'lookup_num': num_reqs,
                           'threshold': threshold})


   if not r:
       # The graph didn't return anything, abandon mission
       st.write('''Your feature description doesn't bring up any relevant ASVS requirements, maybe it doesn't have
                   security impact? Lucky you! Maybe try a lower cut-off, just to be sure''')
       return "Come back with your next feature."


   with st.expander("Here are some requirements for your feature in descending order of relevance, click for details"):
       for entry in r:
           st.write(entry)


   messages = [{
       "role": "system",
       "content": """
           You are an application security specialist. Your role is to assist developers in
           identifying relevant security requirements. You are knowledgeable about OWASP and ASVS in particular
           Attempt to answer the user's question with the context provided.
           Respond in a short, but friendly way.
           Use your knowledge to fill in any gaps, but no hallucinations please.
           If you cannot answer the question, ask for more clarification.
           When formatting your answer, break long text to multiple lines of
           up to 70 characters
           """
   }, {
       "role": "assistant",
       "content": """
               Your Context:
               {}
           """.format(r)
   }, {
       "role": "user",
       "content": """
           Answer the users question, wrapped in three backticks:


           ```
           {}
           ```
           """.format(input_text)
   }]


   with st.spinner("Let me summarise it for you..."):
       # Send the message to OpenAI
       chat_completion = llm.chat.completions.create(model=OPENAI_MODEL, messages=messages)


   return chat_completion.choices[0].message.content




with st.form('my_form'):
   text = st.text_area('Provide a paragraph describing your feature:', 'As a Neo4j PM...')
   submitted = st.form_submit_button('Submit')


   if submitted:
       response = generate_response(text)
       if not response:
           st.write("Something went wrong with OpenAI call.")
       else:
           st.write(response)

실제로 어떤 모습인지 한번 살펴볼까요?

측면 패널은 사용자에게 몇 가지 매개변수와 자신의 환경에서 원하는 키를 가져올 수 있는 기능을 제공해요. 지금은 무시하고 메인 쇼에 집중하면 돼요.

예제의 기능 설명을 사용해서 먼저 ASVS 요구 사항 목록을 얻어볼게요.

의미상 유사한 요구사항 목록(잘림)

이 목록은 컨텍스트로 여러분이 선택한 LLM으로 이동하고, 환각 없이 합리적인 요약으로 보이는 내용을 얻게 돼요.

정말 탄탄한 계획인 것 같네요! 다음은 몇 가지 보안 고려 사항입니다.

1. **데이터 무결성 및 신뢰성**: 상담원의 데이터 보장
BigQuery로 이동하는 것은 암호화되고 인증됩니다. 안전한 사용
HTTPS 또는 TLS와 같은 프로토콜.

2. **속도 제한**: 방지하기 위해 프록시에 속도 제한을 구현합니다.
남용 또는 DoS 공격.

3. **비정상적인 활동에 대한 경고**: ASVS 8.1.4에 따라
시스템은 비정상적인 요청 수를 감지하고 경고할 수 있습니다.
대리인으로부터.

4. **액세스 제어**: 적절한 액세스 제어가 마련되어 있는지 확인하세요.
청구 정보에 액세스하는 상담원과 사용자 모두.
ASVS 4.1.1 지침을 따르십시오.

5. **데이터 개인정보 보호**: 데이터 보호를 준수하는지 확인하세요.
사용자가 GDPR을 내보내거나 삭제할 수 있도록 허용하여 GDPR과 같은 법률을 준수합니다.
데이터(ASVS 8.3.2).

이러한 지침을 따르면 SaaS 제품을 보호하는 데 도움이 됩니다.
새로운 기능을 추가합니다.

두 명의 개발자와 함께 전체 목록을 검토하고 우리 기능에 적용할 수 있는 것이 무엇인지 파악하는 것만큼 포괄적일까요? 물론 그렇지 않아요. 하지만 빠르고 쉽게 시작할 수 있을까요? 보안 요구 사항이 없는 것보다 훨씬 낫겠죠? 전적으로요. 이는 ASVS 및 조직의 전반적인 보안 문화에 대한 추가 참여를 유도할 수 있어요.

물론 모든 기능이 보안 관련성을 갖는 것은 아니에요. 봇이 엔지니어에게 바쁜 작업을 생성하지 않도록 합시다.

네, 괜찮아요.

추가 개발을 위한 아이디어

우리는 이 기본적인 챗봇에 대해 많은 개선 아이디어를 가지고 있어요. 첫째, 모든 사람이 자신의 로컬 컴퓨터에서 실행할 것으로 예상하지 않는 경우 일부 인증을 추가해야 해요. 요구 사항을 백로그 도구로 직접 내보내는 것이 유용한 추가 기능이 될 거예요. 더 야심 찬 기능은 백로그나 설계 문서에서 직접 설명을 가져오는 것이죠.

빠르고 쉬운 방법으로 ASVS 사용을 시작하고 모든 기능에 대해 모든 팀의 전체 요구 사항을 삭제하는 대신 자연스럽게 참여가 증가하도록 하세요. 우리의 접근 방식은 전체 텍스트 검색이나 ASVS에 대한 ChatGPT와의 채팅에 비해 관련 요구 사항을 식별하는 측면에서 훨씬 더 나은 결과를 제공해요 (매우 재미있는 환각이지만).

데이터베이스 덤프 및 코드 다운로드그리고 자신만의 실험을 시작해보세요.

• asvs
• rag

---

에이치시스템즈의 LogTree는 Neo4j 기반 GraphRAG 플랫폼으로, 데이터를 자동으로 지식그래프화하고 자연어 질의로 즉시 답을 제공합니다.

👉 에이치시스템즈 홈페이지